nepi/resources/linux/application.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License version 2 as
   7 #    published by the Free Software Foundation;
   8 #
   9 #    This program is distributed in the hope that it will be useful,
  10 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 #    GNU General Public License for more details.
  13 #
  14 #    You should have received a copy of the GNU General Public License
  15 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  18
  19 from nepi.execution.attribute import Attribute, Flags, Types
  20 from nepi.execution.trace import Trace, TraceAttr
  21 from nepi.execution.resource import ResourceManager, clsinit_copy, \
  22         ResourceState
  23 from nepi.resources.linux.node import LinuxNode
  24 from nepi.util.sshfuncs import ProcStatus, STDOUT
  25 from nepi.util.timefuncs import tnow, tdiffsec
  26
  27 import os
  28 import subprocess
  29
  30 # TODO: Resolve wildcards in commands!!
  31 # TODO: When a failure occurs during deployment, scp and ssh processes are left running behind!!
  32
  33 @clsinit_copy
  34 class LinuxApplication(ResourceManager):
  35     """
  36     .. class:: Class Args :
  37
  38         :param ec: The Experiment controller
  39         :type ec: ExperimentController
  40         :param guid: guid of the RM
  41         :type guid: int
  42
  43     .. note::
  44
  45         A LinuxApplication RM represents a process that can be executed in
  46         a remote Linux host using SSH.
  47
  48         The LinuxApplication RM takes care of uploadin sources and any files
  49         needed to run the experiment, to the remote host.
  50         It also allows to provide source compilation (build) and installation
  51         instructions, and takes care of automating the sources build and
  52         installation tasks for the user.
  53
  54         It is important to note that files uploaded to the remote host have
  55         two possible scopes: single-experiment or multi-experiment.
  56         Single experiment files are those that will not be re-used by other
  57         experiments. Multi-experiment files are those that will.
  58         Sources and shared files are always made available to all experiments.
  59
  60         Directory structure:
  61
  62         The directory structure used by LinuxApplication RM at the Linux
  63         host is the following:
  64
  65         ${HOME}/.nepi/nepi-usr --> Base directory for multi-experiment files
  66                       |
  67         ${LIB}        |- /lib --> Base directory for libraries
  68         ${BIN}        |- /bin --> Base directory for binary files
  69         ${SRC}        |- /src --> Base directory for sources
  70         ${SHARE}      |- /share --> Base directory for other files
  71
  72         ${HOME}/.nepi/nepi-exp --> Base directory for single-experiment files
  73                       |
  74         ${EXP_HOME}   |- /<exp-id>  --> Base directory for experiment exp-id
  75                           |
  76         ${APP_HOME}       |- /<app-guid> --> Base directory for application
  77                                |     specific files (e.g. command.sh, input)
  78                                |
  79         ${RUN_HOME}            |- /<run-id> --> Base directory for run specific
  80
  81     """
  82
  83     _rtype = "linux::Application"
  84     _help = "Runs an application on a Linux host with a BASH command "
  85     _platform = "linux"
  86
  87     @classmethod
  88     def _register_attributes(cls):
  89         cls._register_attribute(
  90             Attribute("command", "Command to execute at application start. "
  91                       "Note that commands will be executed in the ${RUN_HOME} directory, "
  92                       "make sure to take this into account when using relative paths. ",
  93                       flags = Flags.Design))
  94         cls._register_attribute(
  95             Attribute("forwardX11",
  96                       "Enables X11 forwarding for SSH connections",
  97                       flags = Flags.Design))
  98         cls._register_attribute(
  99             Attribute("env",
 100                       "Environment variables string for command execution",
 101                       flags = Flags.Design))
 102         cls._register_attribute(
 103             Attribute("sudo",
 104                       "Run with root privileges",
 105                       flags = Flags.Design))
 106         cls._register_attribute(
 107             Attribute("depends",
 108                       "Space-separated list of packages required to run the application",
 109                       flags = Flags.Design))
 110         cls._register_attribute(
 111             Attribute("sources",
 112                       "semi-colon separated list of regular files to be uploaded to ${SRC} "
 113                       "directory prior to building. Archives won't be expanded automatically. "
 114                       "Sources are globally available for all experiments unless "
 115                       "cleanHome is set to True (This will delete all sources). ",
 116                       flags = Flags.Design))
 117         cls._register_attribute(
 118             Attribute("files",
 119                       "semi-colon separated list of regular miscellaneous files to be uploaded "
 120                       "to ${SHARE} directory. "
 121                       "Files are globally available for all experiments unless "
 122                       "cleanHome is set to True (This will delete all files). ",
 123                       flags = Flags.Design))
 124         cls._register_attribute(
 125             Attribute("libs",
 126                       "semi-colon separated list of libraries (e.g. .so files) to be uploaded "
 127                       "to ${LIB} directory. "
 128                       "Libraries are globally available for all experiments unless "
 129                       "cleanHome is set to True (This will delete all files). ",
 130                       flags = Flags.Design))
 131         cls._register_attribute(
 132             Attribute("bins",
 133                       "semi-colon separated list of binary files to be uploaded "
 134                       "to ${BIN} directory. "
 135                       "Binaries are globally available for all experiments unless "
 136                       "cleanHome is set to True (This will delete all files). ",
 137                       flags = Flags.Design))
 138         cls._register_attribute(
 139             Attribute("code",
 140                       "Plain text source code to be uploaded to the ${APP_HOME} directory. ",
 141                       flags = Flags.Design))
 142         cls._register_attribute(
 143             Attribute("build",
 144                       "Build commands to execute after deploying the sources. "
 145                       "Sources are uploaded to the ${SRC} directory and code "
 146                       "is uploaded to the ${APP_HOME} directory. \n"
 147                       "Usage example: tar xzf ${SRC}/my-app.tgz && cd my-app && "
 148                       "./configure && make && make clean.\n"
 149                       "Make sure to make the build commands return with a nonzero exit "
 150                       "code on error.",
 151                       flags = Flags.Design))
 152         cls._register_attribute(
 153             Attribute("install",
 154                       "Commands to transfer built files to their final destinations. "
 155                       "Install commands are executed after build commands. ",
 156                       flags = Flags.Design))
 157         cls._register_attribute(
 158             Attribute("stdin", "Standard input for the 'command'",
 159                       flags = Flags.Design))
 160         cls._register_attribute(
 161             Attribute("tearDown",
 162                       "Command to be executed just before releasing the resource",
 163                       flags = Flags.Design))
 164         cls._register_attribute(
 165             Attribute("splitStderr",
 166                       "requests stderr to be retrieved separately",
 167                       default = False))
 168
 169     @classmethod
 170     def _register_traces(cls):
 171         stdout = Trace("stdout", "Standard output stream", enabled = True)
 172         stderr = Trace("stderr", "Standard error stream", enabled = True)
 173
 174         cls._register_trace(stdout)
 175         cls._register_trace(stderr)
 176
 177     def __init__(self, ec, guid):
 178         super(LinuxApplication, self).__init__(ec, guid)
 179         self._pid = None
 180         self._ppid = None
 181         self._node = None
 182         self._home = "app-{}".format(self.guid)
 183
 184         # whether the command should run in foreground attached
 185         # to a terminal
 186         self._in_foreground = False
 187
 188         # whether to use sudo to kill the application process
 189         self._sudo_kill = False
 190
 191         # keep a reference to the running process handler when
 192         # the command is not executed as remote daemon in background
 193         self._proc = None
 194
 195         # timestamp of last state check of the application
 196         self._last_state_check = tnow()
 197
 198     def log_message(self, msg):
 199         return " guid {} - host {} - {} "\
 200             .format(self.guid, self.node.get("hostname"), msg)
 201
 202     @property
 203     def node(self):
 204         if not self._node:
 205             node = self.get_connected(LinuxNode.get_rtype())
 206             if not node:
 207                 msg = "Application {} guid {} NOT connected to Node"\
 208                       .format(self._rtype, self.guid)
 209                 raise RuntimeError(msg)
 210
 211             self._node = node[0]
 212
 213         return self._node
 214
 215     @property
 216     def app_home(self):
 217         return os.path.join(self.node.exp_home, self._home)
 218
 219     @property
 220     def run_home(self):
 221         return os.path.join(self.app_home, self.ec.run_id)
 222
 223     @property
 224     def pid(self):
 225         return self._pid
 226
 227     @property
 228     def ppid(self):
 229         return self._ppid
 230
 231     @property
 232     def in_foreground(self):
 233         """
 234         Returns True if the command needs to be executed in foreground.
 235         This means that command will be executed using 'execute' instead of
 236         'run' ('run' executes a command in background and detached from the
 237         terminal)
 238
 239         When using X11 forwarding option, the command can not run in background
 240         and detached from a terminal, since we need to keep the terminal attached
 241         to interact with it.
 242         """
 243         return self.get("forwardX11") or self._in_foreground
 244
 245     def trace_filepath(self, filename):
 246         return os.path.join(self.run_home, filename)
 247
 248     def trace(self, name, attr = TraceAttr.ALL, block = 512, offset = 0):
 249         self.info("Retrieving '{}' trace {} ".format(name, attr))
 250
 251         path = self.trace_filepath(name)
 252
 253         command = "(test -f {} && echo 'success') || echo 'error'".format(path)
 254         (out, err), proc = self.node.execute(command)
 255
 256         if (err and proc.poll()) or out.find("error") != -1:
 257             msg = " Couldn't find trace {} ".format(name)
 258             self.error(msg, out, err)
 259             return None
 260
 261         if attr == TraceAttr.PATH:
 262             return path
 263
 264         if attr == TraceAttr.ALL:
 265             (out, err), proc = self.node.check_output(self.run_home, name)
 266
 267             if proc.poll():
 268                 msg = " Couldn't read trace {} ".format(name)
 269                 self.error(msg, out, err)
 270                 return None
 271
 272             return out
 273
 274         if attr == TraceAttr.STREAM:
 275             cmd = "dd if={} bs={} count=1 skip={}".format(path, block, offset)
 276         elif attr == TraceAttr.SIZE:
 277             cmd = "stat -c {} ".format(path)
 278
 279         (out, err), proc = self.node.execute(cmd)
 280
 281         if proc.poll():
 282             msg = " Couldn't find trace {} ".format(name)
 283             self.error(msg, out, err)
 284             return None
 285
 286         if attr == TraceAttr.SIZE:
 287             out = int(out.strip())
 288
 289         return out
 290
 291     def do_provision(self):
 292         # take a snapshot of the system if user is root
 293         # to ensure that cleanProcess will not kill
 294         # pre-existent processes
 295         if self.node.get("username") == 'root':
 296             import pickle
 297             procs = dict()
 298             ps_aux = "ps aux | awk '{print $2,$11}'"
 299             (out, err), proc = self.node.execute(ps_aux)
 300             if len(out) != 0:
 301                 for line in out.strip().split("\n"):
 302                     parts = line.strip().split(" ")
 303                     procs[parts[0]] = parts[1]
 304                 with open("/tmp/save.proc", "wb") as pickle_file:
 305                     pickle.dump(procs, pickle_file)
 306
 307         # create run dir for application
 308         self.node.mkdir(self.run_home)
 309
 310         # List of all the provision methods to invoke
 311         steps = [
 312             # upload sources
 313             self.upload_sources,
 314             # upload files
 315             self.upload_files,
 316             # upload binaries
 317             self.upload_binaries,
 318             # upload libraries
 319             self.upload_libraries,
 320             # upload code
 321             self.upload_code,
 322             # upload stdin
 323             self.upload_stdin,
 324             # install dependencies
 325             self.install_dependencies,
 326             # build
 327             self.build,
 328             # Install
 329             self.install,
 330         ]
 331
 332         command = []
 333
 334         # Since provisioning takes a long time, before
 335         # each step we check that the EC is still
 336         for step in steps:
 337             if self.ec.abort:
 338                 self.debug("Interrupting provisioning. EC says 'ABORT")
 339                 return
 340
 341             ret = step()
 342             if ret:
 343                 command.append(ret)
 344
 345         # upload deploy script
 346         deploy_command = ";".join(command)
 347         self.execute_deploy_command(deploy_command)
 348
 349         # upload start script
 350         self.upload_start_command()
 351
 352         self.info("Provisioning finished")
 353
 354         super(LinuxApplication, self).do_provision()
 355
 356     def upload_start_command(self, overwrite = False):
 357         # Upload command to remote bash script
 358         # - only if command can be executed in background and detached
 359         command = self.get("command")
 360
 361         if command and not self.in_foreground:
 362 #            self.info("Uploading command '{}'".format(command))
 363
 364             # replace application specific paths in the command
 365             command = self.replace_paths(command)
 366             # replace application specific paths in the environment
 367             env = self.get("env")
 368             env = env and self.replace_paths(env)
 369
 370             shfile = os.path.join(self.app_home, "start.sh")
 371
 372             self.node.upload_command(command,
 373                                      shfile = shfile,
 374                                      env = env,
 375                                      overwrite = overwrite)
 376
 377     def execute_deploy_command(self, command, prefix="deploy"):
 378         if command:
 379             # replace application specific paths in the command
 380             command = self.replace_paths(command)
 381
 382             # replace application specific paths in the environment
 383             env = self.get("env")
 384             env = env and self.replace_paths(env)
 385
 386             # Upload the command to a bash script and run it
 387             # in background ( but wait until the command has
 388             # finished to continue )
 389             shfile = os.path.join(self.app_home, "{}.sh".format(prefix))
 390             # low-level spawn tools in both sshfuncs and execfuncs
 391             # expect stderr=sshfuncs.STDOUT to mean std{out,err} are merged
 392             stderr = "{}_stderr".format(prefix) \
 393                      if self.get("splitStderr") \
 394                         else STDOUT
 395             print("{} : prefix = {}, command={}, stderr={}"
 396                   .format(self, prefix, command, stderr))
 397             self.node.run_and_wait(command, self.run_home,
 398                                    shfile = shfile,
 399                                    overwrite = False,
 400                                    pidfile = "{}_pidfile".format(prefix),
 401                                    ecodefile = "{}_exitcode".format(prefix),
 402                                    stdout = "{}_stdout".format(prefix),
 403                                    stderr = stderr)
 404
 405     def upload_sources(self, sources = None, src_dir = None):
 406         if not sources:
 407             sources = self.get("sources")
 408
 409         command = ""
 410
 411         if not src_dir:
 412             src_dir = self.node.src_dir
 413
 414         if sources:
 415             self.info("Uploading sources ")
 416
 417             sources = [str.strip(source) for source in sources.split(";")]
 418
 419             # Separate sources that should be downloaded from
 420             # the web, from sources that should be uploaded from
 421             # the local machine
 422             command = []
 423             for source in list(sources):
 424                 if source.startswith("http") or source.startswith("https"):
 425                     # remove the hhtp source from the sources list
 426                     sources.remove(source)
 427
 428                     command.append(
 429                         " ( "
 430                         # Check if the source already exists
 431                         " ls {src_dir}/{basename} "
 432                         " || ( "
 433                         # If source doesn't exist, download it and check
 434                         # that it it downloaded ok
 435                         "   wget -c --directory-prefix={src_dir} {source} && "
 436                         "   ls {src_dir}/{basename} "
 437                         " ) ) ".format(
 438                             basename = os.path.basename(source),
 439                             source = source,
 440                             src_dir = src_dir
 441                         ))
 442
 443             command = " && ".join(command)
 444
 445             # replace application specific paths in the command
 446             command = self.replace_paths(command)
 447
 448             if sources:
 449                 sources = ';'.join(sources)
 450                 self.node.upload(sources, src_dir, overwrite = False)
 451
 452         return command
 453
 454     def upload_files(self, files = None):
 455         if not files:
 456             files = self.get("files")
 457
 458         if files:
 459             self.info("Uploading files {} ".format(files))
 460             self.node.upload(files, self.node.share_dir, overwrite = False)
 461
 462     def upload_libraries(self, libs = None):
 463         if not libs:
 464             libs = self.get("libs")
 465
 466         if libs:
 467             self.info("Uploading libraries {} ".format(libs))
 468             self.node.upload(libs, self.node.lib_dir, overwrite = False)
 469
 470     def upload_binaries(self, bins = None):
 471         if not bins:
 472             bins = self.get("bins")
 473
 474         if bins:
 475             self.info("Uploading binaries {} ".format(bins))
 476             self.node.upload(bins, self.node.bin_dir, overwrite = False)
 477
 478     def upload_code(self, code = None):
 479         if not code:
 480             code = self.get("code")
 481
 482         if code:
 483             self.info("Uploading code")
 484             dst = os.path.join(self.app_home, "code")
 485             self.node.upload(code, dst, overwrite = False, text = True, executable = True)
 486
 487     def upload_stdin(self, stdin = None):
 488         if not stdin:
 489            stdin = self.get("stdin")
 490
 491         if stdin:
 492             # create dir for sources
 493             self.info("Uploading stdin")
 494
 495             # upload stdin file to ${SHARE_DIR} directory
 496             if os.path.isfile(stdin):
 497                 basename = os.path.basename(stdin)
 498                 dst = os.path.join(self.node.share_dir, basename)
 499             else:
 500                 dst = os.path.join(self.app_home, "stdin")
 501
 502             self.node.upload(stdin, dst, overwrite = False, text = True)
 503
 504             # create "stdin" symlink on ${APP_HOME} directory
 505             command = "( cd {app_home} ; [ ! -f stdin ] &&  ln -s {stdin} stdin )"\
 506                       .format(app_home = self.app_home, stdin = dst)
 507             return command
 508
 509     def install_dependencies(self, depends = None):
 510         if not depends:
 511             depends = self.get("depends")
 512
 513         if depends:
 514             self.info("Installing dependencies {}".format(depends))
 515             return self.node.install_packages_command(depends)
 516
 517     def build(self, build = None):
 518         if not build:
 519             build = self.get("build")
 520
 521         if build:
 522             self.info("Building sources ")
 523
 524             # replace application specific paths in the command
 525             return self.replace_paths(build)
 526
 527     def install(self, install = None):
 528         if not install:
 529             install = self.get("install")
 530
 531         if install:
 532             self.info("Installing sources ")
 533
 534             # replace application specific paths in the command
 535             return self.replace_paths(install)
 536
 537     def do_deploy(self):
 538         # Wait until node is associated and deployed
 539         node = self.node
 540         if not node or node.state < ResourceState.READY:
 541             self.debug("---- RESCHEDULING DEPLOY ---- node state {} ".format(self.node.state))
 542             self.ec.schedule(self.reschedule_delay, self.deploy)
 543         else:
 544             command = self.get("command") or ""
 545             self.info("Deploying command '{}' ".format(command))
 546             self.do_discover()
 547             self.do_provision()
 548
 549             super(LinuxApplication, self).do_deploy()
 550
 551     def do_start(self):
 552         command = self.get("command")
 553
 554         self.info("Starting command '{}'".format(command))
 555
 556         if not command:
 557             # If no command was given (i.e. Application was used for dependency
 558             # installation), then the application is directly marked as STOPPED
 559             super(LinuxApplication, self).set_stopped()
 560         else:
 561             if self.in_foreground:
 562                 self._run_in_foreground()
 563             else:
 564                 self._run_in_background()
 565
 566             super(LinuxApplication, self).do_start()
 567
 568     def _run_in_foreground(self):
 569         command = self.get("command")
 570         sudo = self.get("sudo") or False
 571         x11 = self.get("forwardX11")
 572         env = self.get("env")
 573
 574         # Command will be launched in foreground and attached to the
 575         # terminal using the node 'execute' in non blocking mode.
 576
 577         # We save the reference to the process in self._proc
 578         # to be able to kill the process from the stop method.
 579         # We also set blocking = False, since we don't want the
 580         # thread to block until the execution finishes.
 581         (out, err), self._proc = self.execute_command(command,
 582                                                       env = env,
 583                                                       sudo = sudo,
 584                                                       forward_x11 = x11,
 585                                                       blocking = False)
 586
 587         if self._proc.poll():
 588             self.error(msg, out, err)
 589             raise RuntimeError(msg)
 590
 591     def _run_in_background(self):
 592         command = self.get("command")
 593         env = self.get("env")
 594         sudo = self.get("sudo") or False
 595
 596         stdout = "stdout"
 597         # low-level spawn tools in both sshfuncs and execfuncs
 598         # expect stderr=sshfuncs.STDOUT to mean std{out,err} are merged
 599         stderr = "stderr" \
 600                  if self.get("splitStderr") \
 601                     else STDOUT
 602         stdin = os.path.join(self.app_home, "stdin") if self.get("stdin") \
 603                 else None
 604
 605         # Command will be run as a daemon in baground and detached from any
 606         # terminal.
 607         # The command to run was previously uploaded to a bash script
 608         # during deployment, now we launch the remote script using 'run'
 609         # method from the node.
 610         cmd = "bash {}".format(os.path.join(self.app_home, "start.sh"))
 611         (out, err), proc = self.node.run(cmd, self.run_home,
 612                                          stdin = stdin,
 613                                          stdout = stdout,
 614                                          stderr = stderr,
 615                                          sudo = sudo)
 616
 617         # check if execution errors occurred
 618         msg = " Failed to start command '{}' ".format(command)
 619
 620         if proc.poll():
 621             self.error(msg, out, err)
 622             raise RuntimeError(msg)
 623
 624         # Wait for pid file to be generated
 625         pid, ppid = self.node.wait_pid(self.run_home)
 626         if pid: self._pid = int(pid)
 627         if ppid: self._ppid = int(ppid)
 628
 629         # If the process is not running, check for error information
 630         # on the remote machine
 631         if not self.pid or not self.ppid:
 632             (out, err), proc = self.node.check_errors(self.run_home,
 633                                                       stderr = stderr)
 634
 635             # Out is what was written in the stderr file
 636             if err:
 637                 msg = " Failed to start command '{}' ".format(command)
 638                 self.error(msg, out, err)
 639                 raise RuntimeError(msg)
 640
 641     def do_stop(self):
 642         """ Stops application execution
 643         """
 644         command = self.get('command') or ''
 645
 646         if self.state == ResourceState.STARTED:
 647
 648             self.info("Stopping command '{}' ".format(command))
 649
 650             # If the command is running in foreground (it was launched using
 651             # the node 'execute' method), then we use the handler to the Popen
 652             # process to kill it. Else we send a kill signal using the pid and ppid
 653             # retrieved after running the command with the node 'run' method
 654             if self._proc:
 655                 self._proc.kill()
 656             else:
 657                 # Only try to kill the process if the pid and ppid
 658                 # were retrieved
 659                 if self.pid and self.ppid:
 660                     (out, err), proc = self.node.kill(self.pid, self.ppid,
 661                                                       sudo = self._sudo_kill)
 662
 663                     """
 664                     # TODO: check if execution errors occurred
 665                     if (proc and proc.poll()) or err:
 666                         msg = " Failed to STOP command '{}' ".format(self.get("command"))
 667                         self.error(msg, out, err)
 668                     """
 669
 670             super(LinuxApplication, self).do_stop()
 671
 672     def do_release(self):
 673         self.info("Releasing resource")
 674
 675         self.do_stop()
 676
 677         tear_down = self.get("tearDown")
 678         if tear_down:
 679             self.node.execute(tear_down)
 680
 681         hard_release = self.get("hardRelease")
 682         if hard_release:
 683             self.node.rmdir(self.app_home)
 684
 685         super(LinuxApplication, self).do_release()
 686
 687     @property
 688     def state(self):
 689         """ Returns the state of the application
 690         """
 691         if self._state == ResourceState.STARTED:
 692             if self.in_foreground:
 693                 # Check if the process we used to execute the command
 694                 # is still running ...
 695                 retcode = self._proc.poll()
 696
 697                 # retcode == None -> running
 698                 # retcode > 0 -> error
 699                 # retcode == 0 -> finished
 700                 if retcode:
 701                     out = ""
 702                     msg = " Failed to execute command '{}'".format(self.get("command"))
 703                     err = self._proc.stderr.read()
 704                     self.error(msg, out, err)
 705                     self.do_fail()
 706
 707                 elif retcode == 0:
 708                     self.set_stopped()
 709             else:
 710                 # We need to query the status of the command we launched in
 711                 # background. In order to avoid overwhelming the remote host and
 712                 # the local processor with too many ssh queries, the state is only
 713                 # requested every 'state_check_delay' seconds.
 714                 state_check_delay = 0.5
 715                 if tdiffsec(tnow(), self._last_state_check) > state_check_delay:
 716                     if self.pid and self.ppid:
 717                         # Make sure the process is still running in background
 718                         status = self.node.status(self.pid, self.ppid)
 719
 720                         if status == ProcStatus.FINISHED:
 721                             # If the program finished, check if execution
 722                             # errors occurred
 723                             (out, err), proc \
 724                                 = self.node.check_errors(self.run_home)
 725
 726                             if err:
 727                                 msg = "Failed to execute command '{}'"\
 728                                       .format(self.get("command"))
 729                                 self.error(msg, out, err)
 730                                 self.do_fail()
 731                             else:
 732                                 self.set_stopped()
 733
 734                     self._last_state_check = tnow()
 735
 736         return self._state
 737
 738     def execute_command(self, command,
 739                         env = None,
 740                         sudo = False,
 741                         tty = False,
 742                         forward_x11 = False,
 743                         blocking = False):
 744
 745         environ = ""
 746         if env:
 747             environ = self.node.format_environment(env, inline = True)
 748         command = environ + command
 749         command = self.replace_paths(command)
 750
 751         return self.node.execute(command,
 752                                  sudo = sudo,
 753                                  tty = tty,
 754                                  forward_x11 = forward_x11,
 755                                  blocking = blocking)
 756
 757     def replace_paths(self, command, node = None, app_home = None, run_home = None):
 758         """
 759         Replace all special path tags with shell-escaped actual paths.
 760         """
 761         if not node:
 762             node = self.node
 763
 764         if not app_home:
 765             app_home = self.app_home
 766
 767         if not run_home:
 768             run_home = self.run_home
 769
 770         return ( command
 771                  .replace("${USR}", node.usr_dir)
 772                  .replace("${LIB}", node.lib_dir)
 773                  .replace("${BIN}", node.bin_dir)
 774                  .replace("${SRC}", node.src_dir)
 775                  .replace("${SHARE}", node.share_dir)
 776                  .replace("${EXP}", node.exp_dir)
 777                  .replace("${EXP_HOME}", node.exp_home)
 778                  .replace("${APP_HOME}", app_home)
 779                  .replace("${RUN_HOME}", run_home)
 780                  .replace("${NODE_HOME}", node.node_home)
 781                  .replace("${HOME}", node.home_dir)
 782                  # a shortcut to refer to the file uploaded as 'code = '
 783                  .replace("${CODE}", "{}/code".format(app_home))
 784              )
 785
 786     def valid_connection(self, guid):
 787         # TODO: Validate!
 788         return True