src/nepi/testbeds/planetlab/application.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 from constants import TESTBED_ID
   5 import plcapi
   6 import operator
   7 import os
   8 import os.path
   9 import nepi.util.server as server
  10 import cStringIO
  11 import subprocess
  12 import rspawn
  13
  14 from nepi.util.constants import STATUS_NOT_STARTED, STATUS_RUNNING, \
  15         STATUS_FINISHED
  16
  17 class Dependency(object):
  18     """
  19     A Dependency is in every respect like an application.
  20
  21     It depends on some packages, it may require building binaries, it must deploy
  22     them...
  23
  24     But it has no command. Dependencies aren't ever started, or stopped, and have
  25     no status.
  26     """
  27
  28     TRACES = ('buildlog')
  29
  30     def __init__(self, api=None):
  31         if not api:
  32             api = plcapi.PLCAPI()
  33         self._api = api
  34
  35         # Attributes
  36         self.command = None
  37         self.sudo = False
  38
  39         self.build = None
  40         self.install = None
  41         self.depends = None
  42         self.buildDepends = None
  43         self.sources = None
  44         self.env = {}
  45
  46         self.stdin = None
  47         self.stdout = None
  48         self.stderr = None
  49         self.buildlog = None
  50
  51         self.add_to_path = True
  52
  53         # Those are filled when the app is configured
  54         self.home_path = None
  55
  56         # Those are filled when an actual node is connected
  57         self.node = None
  58
  59         # Those are filled when the app is started
  60         #   Having both pid and ppid makes it harder
  61         #   for pid rollover to induce tracking mistakes
  62         self._started = False
  63         self._setup = False
  64         self._setuper = None
  65         self._pid = None
  66         self._ppid = None
  67
  68     def __str__(self):
  69         return "%s<%s>" % (
  70             self.__class__.__name__,
  71             ' '.join(list(self.depends or [])
  72                    + list(self.sources or []))
  73         )
  74
  75     def validate(self):
  76         if self.home_path is None:
  77             raise AssertionError, "Misconfigured application: missing home path"
  78         if self.node.ident_path is None or not os.access(self.node.ident_path, os.R_OK):
  79             raise AssertionError, "Misconfigured application: missing slice SSH key"
  80         if self.node is None:
  81             raise AssertionError, "Misconfigured application: unconnected node"
  82         if self.node.hostname is None:
  83             raise AssertionError, "Misconfigured application: misconfigured node"
  84         if self.node.slicename is None:
  85             raise AssertionError, "Misconfigured application: unspecified slice"
  86
  87     def remote_trace_path(self, whichtrace):
  88         if whichtrace in self.TRACES:
  89             tracefile = os.path.join(self.home_path, whichtrace)
  90         else:
  91             tracefile = None
  92
  93         return tracefile
  94
  95     def sync_trace(self, local_dir, whichtrace):
  96         tracefile = self.remote_trace_path(whichtrace)
  97         if not tracefile:
  98             return None
  99
 100         local_path = os.path.join(local_dir, tracefile)
 101
 102         # create parent local folders
 103         proc = subprocess.Popen(
 104             ["mkdir", "-p", os.path.dirname(local_path)],
 105             stdout = open("/dev/null","w"),
 106             stdin = open("/dev/null","r"))
 107
 108         if proc.wait():
 109             raise RuntimeError, "Failed to synchronize trace"
 110
 111         # sync files
 112         (out,err),proc = server.popen_scp(
 113             '%s@%s:%s' % (self.node.slicename, self.node.hostname,
 114                 tracefile),
 115             local_path,
 116             port = None,
 117             agent = None,
 118             ident_key = self.node.ident_path,
 119             server_key = self.node.server_key
 120             )
 121
 122         if proc.wait():
 123             raise RuntimeError, "Failed to synchronize trace: %s %s" % (out,err,)
 124
 125         return local_path
 126
 127
 128     def setup(self):
 129         self._make_home()
 130         self._build()
 131         self._setup = True
 132
 133     def async_setup(self):
 134         if not self._setuper:
 135             self._setuper = threading.Thread(
 136                 target = self.setup)
 137             self._setuper.start()
 138
 139     def async_setup_wait(self):
 140         if not self._setup:
 141             if self._setuper:
 142                 self._setuper.join()
 143                 if not self._setup:
 144                     raise RuntimeError, "Failed to setup application"
 145             else:
 146                 self.setup()
 147
 148     def _make_home(self):
 149         # Make sure all the paths are created where
 150         # they have to be created for deployment
 151         (out,err),proc = server.popen_ssh_command(
 152             "mkdir -p %s" % (server.shell_escape(self.home_path),),
 153             host = self.node.hostname,
 154             port = None,
 155             user = self.node.slicename,
 156             agent = None,
 157             ident_key = self.node.ident_path,
 158             server_key = self.node.server_key
 159             )
 160
 161         if proc.wait():
 162             raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
 163
 164
 165         if self.stdin:
 166             # Write program input
 167             (out,err),proc = server.popen_scp(
 168                 cStringIO.StringIO(self.stdin),
 169                 '%s@%s:%s' % (self.node.slicename, self.node.hostname,
 170                     os.path.join(self.home_path, 'stdin') ),
 171                 port = None,
 172                 agent = None,
 173                 ident_key = self.node.ident_path,
 174                 server_key = self.node.server_key
 175                 )
 176
 177             if proc.wait():
 178                 raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
 179
 180     def _replace_paths(self, command):
 181         """
 182         Replace all special path tags with shell-escaped actual paths.
 183         """
 184         # need to append ${HOME} if paths aren't absolute, to MAKE them absolute.
 185         root = '' if self.home_path.startswith('/') else "${HOME}/"
 186         return ( command
 187             .replace("${SOURCES}", root+server.shell_escape(self.home_path))
 188             .replace("${BUILD}", root+server.shell_escape(os.path.join(self.home_path,'build'))) )
 189
 190     def _build(self):
 191         if self.sources:
 192             sources = self.sources.split(' ')
 193
 194             # Copy all sources
 195             (out,err),proc = server.popen_scp(
 196                 sources,
 197                 "%s@%s:%s" % (self.node.slicename, self.node.hostname,
 198                     os.path.join(self.home_path,'.'),),
 199                 ident_key = self.node.ident_path,
 200                 server_key = self.node.server_key
 201                 )
 202
 203             if proc.wait():
 204                 raise RuntimeError, "Failed upload source file %r: %s %s" % (source, out,err,)
 205
 206         if self.buildDepends:
 207             # Install build dependencies
 208             (out,err),proc = server.popen_ssh_command(
 209                 "sudo -S yum -y install %(packages)s" % {
 210                     'packages' : self.buildDepends
 211                 },
 212                 host = self.node.hostname,
 213                 port = None,
 214                 user = self.node.slicename,
 215                 agent = None,
 216                 ident_key = self.node.ident_path,
 217                 server_key = self.node.server_key
 218                 )
 219
 220             if proc.wait():
 221                 raise RuntimeError, "Failed instal build dependencies: %s %s" % (out,err,)
 222
 223
 224         if self.build:
 225             # Build sources
 226             (out,err),proc = server.popen_ssh_command(
 227                 "cd %(home)s && mkdir -p build && cd build && ( %(command)s ) > ${HOME}/%(home)s/buildlog 2>&1 || ( tail ${HOME}/%(home)s/buildlog >&2 && false )" % {
 228                     'command' : self._replace_paths(self.build),
 229                     'home' : server.shell_escape(self.home_path),
 230                 },
 231                 host = self.node.hostname,
 232                 port = None,
 233                 user = self.node.slicename,
 234                 agent = None,
 235                 ident_key = self.node.ident_path,
 236                 server_key = self.node.server_key
 237                 )
 238
 239             if proc.wait():
 240                 raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
 241
 242             # Make archive
 243             (out,err),proc = server.popen_ssh_command(
 244                 "cd %(home)s && tar czf build.tar.gz build" % {
 245                     'command' : self._replace_paths(self.build),
 246                     'home' : server.shell_escape(self.home_path),
 247                 },
 248                 host = self.node.hostname,
 249                 port = None,
 250                 user = self.node.slicename,
 251                 agent = None,
 252                 ident_key = self.node.ident_path,
 253                 server_key = self.node.server_key
 254                 )
 255
 256             if proc.wait():
 257                 raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
 258
 259         if self.install:
 260             # Install application
 261             (out,err),proc = server.popen_ssh_command(
 262                 "cd %(home)s && cd build && ( %(command)s ) > ${HOME}/%(home)s/installlog 2>&1 || ( tail ${HOME}/%(home)s/installlog >&2 && false )" % {
 263                     'command' : self._replace_paths(self.install),
 264                     'home' : server.shell_escape(self.home_path),
 265                 },
 266                 host = self.node.hostname,
 267                 port = None,
 268                 user = self.node.slicename,
 269                 agent = None,
 270                 ident_key = self.node.ident_path,
 271                 server_key = self.node.server_key
 272                 )
 273
 274             if proc.wait():
 275                 raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
 276
 277 class Application(Dependency):
 278     """
 279     An application also has dependencies, but also a command to be ran and monitored.
 280
 281     It adds the output of that command as traces.
 282     """
 283
 284     TRACES = ('stdout','stderr','buildlog')
 285
 286     def __init__(self, api=None):
 287         super(Application,self).__init__(api)
 288
 289         # Attributes
 290         self.command = None
 291         self.sudo = False
 292
 293         self.stdin = None
 294         self.stdout = None
 295         self.stderr = None
 296
 297         # Those are filled when the app is started
 298         #   Having both pid and ppid makes it harder
 299         #   for pid rollover to induce tracking mistakes
 300         self._started = False
 301         self._pid = None
 302         self._ppid = None
 303
 304         # Do not add to the python path of nodes
 305         self.add_to_path = False
 306
 307     def __str__(self):
 308         return "%s<command:%s%s>" % (
 309             self.__class__.__name__,
 310             "sudo " if self.sudo else "",
 311             self.command,
 312         )
 313
 314     def start(self):
 315         # Create shell script with the command
 316         # This way, complex commands and scripts can be ran seamlessly
 317         # sync files
 318         command = cStringIO.StringIO()
 319         command.write('export PYTHONPATH=$PYTHONPATH:%s\n' % (
 320             ':'.join(["${HOME}/"+server.shell_escape(s) for s in self.node.pythonpath])
 321         ))
 322         command.write('export PATH=$PATH:%s\n' % (
 323             ':'.join(["${HOME}/"+server.shell_escape(s) for s in self.node.pythonpath])
 324         ))
 325         if self.node.env:
 326             for envkey, envvals in self.node.env.iteritems():
 327                 for envval in envvals:
 328                     command.write('export %s=%s\n' % (envkey, envval))
 329         command.write(self.command)
 330         command.seek(0)
 331
 332         (out,err),proc = server.popen_scp(
 333             command,
 334             '%s@%s:%s' % (self.node.slicename, self.node.hostname,
 335                 os.path.join(self.home_path, "app.sh")),
 336             port = None,
 337             agent = None,
 338             ident_key = self.node.ident_path,
 339             server_key = self.node.server_key
 340             )
 341
 342         if proc.wait():
 343             raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
 344
 345         # Start process in a "daemonized" way, using nohup and heavy
 346         # stdin/out redirection to avoid connection issues
 347         (out,err),proc = rspawn.remote_spawn(
 348             self._replace_paths("bash ./app.sh"),
 349
 350             pidfile = './pid',
 351             home = self.home_path,
 352             stdin = 'stdin' if self.stdin is not None else '/dev/null',
 353             stdout = 'stdout' if self.stdout else '/dev/null',
 354             stderr = 'stderr' if self.stderr else '/dev/null',
 355             sudo = self.sudo,
 356
 357             host = self.node.hostname,
 358             port = None,
 359             user = self.node.slicename,
 360             agent = None,
 361             ident_key = self.node.ident_path,
 362             server_key = self.node.server_key
 363             )
 364
 365         if proc.wait():
 366             raise RuntimeError, "Failed to set up application: %s %s" % (out,err,)
 367
 368         self._started = True
 369
 370     def checkpid(self):
 371         # Get PID/PPID
 372         # NOTE: wait a bit for the pidfile to be created
 373         if self._started and not self._pid or not self._ppid:
 374             pidtuple = rspawn.remote_check_pid(
 375                 os.path.join(self.home_path,'pid'),
 376                 host = self.node.hostname,
 377                 port = None,
 378                 user = self.node.slicename,
 379                 agent = None,
 380                 ident_key = self.node.ident_path,
 381                 server_key = self.node.server_key
 382                 )
 383
 384             if pidtuple:
 385                 self._pid, self._ppid = pidtuple
 386
 387     def status(self):
 388         self.checkpid()
 389         if not self._started:
 390             return STATUS_NOT_STARTED
 391         elif not self._pid or not self._ppid:
 392             return STATUS_NOT_STARTED
 393         else:
 394             status = rspawn.remote_status(
 395                 self._pid, self._ppid,
 396                 host = self.node.hostname,
 397                 port = None,
 398                 user = self.node.slicename,
 399                 agent = None,
 400                 ident_key = self.node.ident_path
 401                 )
 402
 403             if status is rspawn.NOT_STARTED:
 404                 return STATUS_NOT_STARTED
 405             elif status is rspawn.RUNNING:
 406                 return STATUS_RUNNING
 407             elif status is rspawn.FINISHED:
 408                 return STATUS_FINISHED
 409             else:
 410                 # WTF?
 411                 return STATUS_NOT_STARTED
 412
 413     def kill(self):
 414         status = self.status()
 415         if status == STATUS_RUNNING:
 416             # kill by ppid+pid - SIGTERM first, then try SIGKILL
 417             rspawn.remote_kill(
 418                 self._pid, self._ppid,
 419                 host = self.node.hostname,
 420                 port = None,
 421                 user = self.node.slicename,
 422                 agent = None,
 423                 ident_key = self.node.ident_path,
 424                 server_key = self.node.server_key
 425                 )
 426
 427 class NepiDependency(Dependency):
 428     """
 429     This dependency adds nepi itself to the python path,
 430     so that you may run testbeds within PL nodes.
 431     """
 432
 433     # Class attribute holding a *weak* reference to the shared NEPI tar file
 434     # so that they may share it. Don't operate on the file itself, it would
 435     # be a mess, just use its path.
 436     _shared_nepi_tar = None
 437
 438     def __init__(self, api = None):
 439         super(NepiDependency, self).__init__(api)
 440
 441         self._tarball = None
 442
 443         self.depends = 'python python-ipaddr python-setuptools'
 444
 445         # our sources are in our ad-hoc tarball
 446         self.sources = self.tarball.name
 447
 448         tarname = os.path.basename(self.tarball.name)
 449
 450         # it's already built - just move the tarball into place
 451         self.build = "mv ${SOURCES}/%s ." % (tarname,)
 452
 453         # unpack it into sources, and we're done
 454         self.install = "tar xzf ${BUILD}/%s -C .." % (tarname,)
 455
 456     @property
 457     def tarball(self):
 458         if self._tarball is None:
 459             shared_tar = self._shared_nepi_tar and self._shared_nepi_tar()
 460             if shared_tar is not None:
 461                 self._tarball = shared_tar
 462             else:
 463                 # Build an ad-hoc tarball
 464                 # Prebuilt
 465                 import nepi
 466                 import tempfile
 467
 468                 shared_tar = tempfile.NamedTemporaryFile(prefix='nepi-src-', suffix='.tar.gz')
 469
 470                 proc = subprocess.Popen(
 471                     ["tar", "czf", shared_tar.name,
 472                         '-C', os.path.join(os.path.dirname(os.path.dirname(nepi.__file__)),'.'),
 473                         'nepi'],
 474                     stdout = open("/dev/null","w"),
 475                     stdin = open("/dev/null","r"))
 476
 477                 if proc.wait():
 478                     raise RuntimeError, "Failed to create nepi tarball"
 479
 480                 self._tarball = self._shared_nepi_tar = shared_tar
 481
 482         return self._tarball
 483
 484 class NS3Dependency(Dependency):
 485     """
 486     This dependency adds NS3 libraries to the library paths,
 487     so that you may run the NS3 testbed within PL nodes.
 488
 489     You'll also need the NepiDependency.
 490     """
 491
 492     def __init__(self, api = None):
 493         super(NS3Dependency, self).__init__(api)
 494
 495         self.buildDepends = 'build-essential waf gcc gcc-c++ gccxml unzip'
 496
 497         # We have to download the sources, untar, build...
 498         pybindgen_source_url = "http://pybindgen.googlecode.com/files/pybindgen-0.15.0.zip"
 499         pygccxml_source_url = "http://leaseweb.dl.sourceforge.net/project/pygccxml/pygccxml/pygccxml-1.0/pygccxml-1.0.0.zip"
 500         ns3_source_url = "http://yans.pl.sophia.inria.fr/code/hgwebdir.cgi/ns-3-dev/archive/tip.tar.gz"
 501         self.build =(
 502             " ( "
 503             "  cd .. && "
 504             "  python -c 'import pygccxml, pybindgen' && "
 505             "  test -f lib/_ns3.so && "
 506             "  test -f lib/libns3.so "
 507             " ) || ( "
 508                 # Not working, rebuild
 509                      "wget -q -c -O pybindgen-src.zip %(pybindgen_source_url)s && " # continue, to exploit the case when it has already been dl'ed
 510                      "wget -q -c -O pygccxml-1.0.0.zip %(pygccxml_source_url)s && "
 511                      "wget -q -c -O ns3-src.tar.gz %(ns3_source_url)s && "
 512                      "unzip -n pybindgen-src.zip && " # Do not overwrite files, to exploit the case when it has already been built
 513                      "unzip -n pygccxml-1.0.0.zip && "
 514                      "mkdir -p ns3-src && "
 515                      "tar xzf ns3-src.tar.gz --strip-components=1 -C ns3-src && "
 516                      "rm -rf target && "    # mv doesn't like unclean targets
 517                      "mkdir -p target && "
 518                      "cd pygccxml-1.0.0 && "
 519                      "rm -rf unittests docs && " # pygccxml has ~100M of unit tests - excessive - docs aren't needed either
 520                      "python setup.py build && "
 521                      "python setup.py install --install-lib ${BUILD}/target && "
 522                      "python setup.py clean && "
 523                      "cd ../pybindgen-0.15.0 && "
 524                      "export PYTHONPATH=$PYTHONPATH:${BUILD}/target && "
 525                      "./waf configure --prefix=${BUILD}/target -d release && "
 526                      "./waf && "
 527                      "./waf install && "
 528                      "./waf clean && "
 529                      "mv -f ${BUILD}/target/lib/python*/site-packages/pybindgen ${BUILD}/target/. && "
 530                      "rm -rf ${BUILD}/target/lib && "
 531                      "cd ../ns3-src && "
 532                      "./waf configure --prefix=${BUILD}/target -d release --disable-examples --high-precision-as-double && "
 533                      "./waf &&"
 534                      "./waf install && "
 535                      "./waf clean"
 536              " )"
 537                      % dict(
 538                         pybindgen_source_url = server.shell_escape(pybindgen_source_url),
 539                         pygccxml_source_url = server.shell_escape(pygccxml_source_url),
 540                         ns3_source_url = server.shell_escape(ns3_source_url),
 541                      ))
 542
 543         # Just move ${BUILD}/target
 544         self.install = (
 545             " ( "
 546             "  cd .. && "
 547             "  python -c 'import pygccxml, pybindgen' && "
 548             "  test -f lib/_ns3.so && "
 549             "  test -f lib/libns3.so "
 550             " ) || ( "
 551                 # Not working, reinstall
 552                     "( for i in ${BUILD}/target/* ; do rm -rf ${SOURCES}/${i##*/} ; done ) && " # mv doesn't like unclean targets
 553                     "mv -f ${BUILD}/target/* ${SOURCES}"
 554             " )"
 555         )
 556
 557         # Set extra environment paths
 558         self.env['NEPI_NS3BINDINGS'] = "${SOURCES}/lib"
 559         self.env['NEPI_NS3LIBRARY'] = "${SOURCES}/lib/libns3.so"
 560
 561     @property
 562     def tarball(self):
 563         if self._tarball is None:
 564             shared_tar = self._shared_nepi_tar and self._shared_nepi_tar()
 565             if shared_tar is not None:
 566                 self._tarball = shared_tar
 567             else:
 568                 # Build an ad-hoc tarball
 569                 # Prebuilt
 570                 import nepi
 571                 import tempfile
 572
 573                 shared_tar = tempfile.NamedTemporaryFile(prefix='nepi-src-', suffix='.tar.gz')
 574
 575                 proc = subprocess.Popen(
 576                     ["tar", "czf", shared_tar.name,
 577                         '-C', os.path.join(os.path.dirname(os.path.dirname(nepi.__file__)),'.'),
 578                         'nepi'],
 579                     stdout = open("/dev/null","w"),
 580                     stdin = open("/dev/null","r"))
 581
 582                 if proc.wait():
 583                     raise RuntimeError, "Failed to create nepi tarball"
 584
 585                 self._tarball = self._shared_nepi_tar = shared_tar
 586
 587         return self._tarball
 588
 589