src/neco/resources/linux/node.py

   1 from neco.execution.attribute import Attribute, Flags
   2 from neco.execution.resource import ResourceManager, clsinit, ResourceState
   3 from neco.resources.linux import rpmfuncs, debfuncs
   4 from neco.util import sshfuncs, execfuncs
   5
   6 import collections
   7 import logging
   8 import os
   9 import random
  10 import re
  11 import tempfile
  12 import time
  13 import threading
  14
  15 # TODO: Verify files and dirs exists already
  16
  17 @clsinit
  18 class LinuxNode(ResourceManager):
  19     _rtype = "LinuxNode"
  20
  21     @classmethod
  22     def _register_attributes(cls):
  23         hostname = Attribute("hostname", "Hostname of the machine")
  24
  25         username = Attribute("username", "Local account username",
  26                 flags = Flags.Credential)
  27
  28         port = Attribute("port", "SSH port", flags = Flags.Credential)
  29
  30         home = Attribute("home",
  31                 "Experiment home directory to store all experiment related files")
  32
  33         identity = Attribute("identity", "SSH identity file",
  34                 flags = Flags.Credential)
  35
  36         server_key = Attribute("serverKey", "Server public key",
  37                 flags = Flags.Credential)
  38
  39         clean_home = Attribute("cleanHome", "Remove all files and directories " + \
  40                 " from home folder before starting experiment",
  41                 flags = Flags.ReadOnly)
  42
  43         clean_processes = Attribute("cleanProcesses",
  44                 "Kill all running processes before starting experiment",
  45                 flags = Flags.ReadOnly)
  46
  47         tear_down = Attribute("tearDown", "Bash script to be executed before " + \
  48                 "releasing the resource", flags = Flags.ReadOnly)
  49
  50         cls._register_attribute(hostname)
  51         cls._register_attribute(username)
  52         cls._register_attribute(port)
  53         cls._register_attribute(home)
  54         cls._register_attribute(identity)
  55         cls._register_attribute(server_key)
  56         cls._register_attribute(clean_home)
  57         cls._register_attribute(clean_processes)
  58         cls._register_attribute(tear_down)
  59
  60     def __init__(self, ec, guid):
  61         super(LinuxNode, self).__init__(ec, guid)
  62         self._os = None
  63         self._home = "nepi-exp-%s" % os.urandom(8).encode('hex')
  64
  65         # lock to avoid concurrency issues on methods used by applications
  66         self._lock = threading.Lock()
  67
  68         self._logger = logging.getLogger("neco.linux.Node.%d " % self.guid)
  69
  70     @property
  71     def home(self):
  72         home = self.get("home")
  73         if home and not home.startswith("nepi-"):
  74             home = "nepi-" + home
  75         return home or self._home
  76
  77     @property
  78     def os(self):
  79         if self._os:
  80             return self._os
  81
  82         if (not self.get("hostname") or not self.get("username")):
  83             msg = "Can't resolve OS for guid %d. Insufficient data." % self.guid
  84             self.logger.error(msg)
  85             raise RuntimeError, msg
  86
  87         (out, err), proc = self.execute("cat /etc/issue")
  88
  89         if err and proc.poll():
  90             msg = "Error detecting OS for host %s. err: %s " % (self.get("hostname"), err)
  91             self.logger.error(msg)
  92             raise RuntimeError, msg
  93
  94         if out.find("Fedora release 12") == 0:
  95             self._os = "f12"
  96         elif out.find("Fedora release 14") == 0:
  97             self._os = "f14"
  98         elif out.find("Debian") == 0:
  99             self._os = "debian"
 100         elif out.find("Ubuntu") ==0:
 101             self._os = "ubuntu"
 102         else:
 103             msg = "Unsupported OS %s for host %s" % (out, self.get("hostname"))
 104             self.logger.error(msg)
 105             raise RuntimeError, msg
 106
 107         return self._os
 108
 109     @property
 110     def localhost(self):
 111         return self.get("hostname") in ['localhost', '127.0.0.7', '::1']
 112
 113     def provision(self, filters = None):
 114         if not self.is_alive():
 115             self._state = ResourceState.FAILED
 116             self.logger.error("Deploy failed. Unresponsive node")
 117             return
 118
 119     def deploy(self):
 120         self.provision()
 121
 122         if self.get("cleanProcesses"):
 123             self.clean_processes()
 124
 125         if self.get("cleanHome"):
 126             # self.clean_home() -> this is dangerous
 127             pass
 128
 129         self.mkdir(self.home)
 130
 131         super(LinuxNode, self).deploy()
 132
 133     def release(self):
 134         tear_down = self.get("tearDown")
 135         if tear_down:
 136             self.execute(tear_down)
 137
 138         super(LinuxNode, self).release()
 139
 140     def validate_connection(self, guid):
 141         # TODO: Validate!
 142         return True
 143
 144     def clean_processes(self):
 145         self.logger.info("Cleaning up processes")
 146
 147         cmd = ("sudo -S killall python tcpdump || /bin/true ; " +
 148             "sudo -S killall python tcpdump || /bin/true ; " +
 149             "sudo -S kill $(ps -N -T -o pid --no-heading | grep -v $PPID | sort) || /bin/true ; " +
 150             "sudo -S killall -u root || /bin/true ; " +
 151             "sudo -S killall -u root || /bin/true ; ")
 152
 153         out = err = ""
 154         with self._lock:
 155            (out, err), proc = self.run_and_wait(cmd, self.home,
 156                 pidfile = "cppid",
 157                 stdout = "cplog",
 158                 stderr = "cperr",
 159                 raise_on_error = True)
 160
 161         return (out, err)
 162
 163     def clean_home(self):
 164         self.logger.info("Cleaning up home")
 165
 166         cmd = "find . -maxdepth 1  \( -name '.cache' -o -name '.local' -o -name '.config' -o -name 'nepi-*' \) -execdir rm -rf {} + "
 167
 168         out = err = ""
 169         with self._lock:
 170             (out, err), proc = self.run_and_wait(cmd, self.home,
 171                 pidfile = "chpid",
 172                 stdout = "chlog",
 173                 stderr = "cherr",
 174                 raise_on_error = True)
 175
 176         return (out, err)
 177
 178     def upload(self, src, dst):
 179         """ Copy content to destination
 180
 181            src  content to copy. Can be a local file, directory or text input
 182
 183            dst  destination path on the remote host (remote is always self.host)
 184         """
 185         # If source is a string input
 186         if not os.path.isfile(src):
 187             # src is text input that should be uploaded as file
 188             # create a temporal file with the content to upload
 189             f = tempfile.NamedTemporaryFile(delete=False)
 190             f.write(src)
 191             f.close()
 192             src = f.name
 193
 194         if not self.localhost:
 195             # Build destination as <user>@<server>:<path>
 196             dst = "%s@%s:%s" % (self.get("username"), self.get("hostname"), dst)
 197
 198         return self.copy(src, dst)
 199
 200     def download(self, src, dst):
 201         if not self.localhost:
 202             # Build destination as <user>@<server>:<path>
 203             src = "%s@%s:%s" % (self.get("username"), self.get("hostname"), src)
 204         return self.copy(src, dst)
 205
 206     def install_packages(self, packages):
 207         cmd = ""
 208         if self.os in ["f12", "f14"]:
 209             cmd = rpmfuncs.install_packages_command(self.os, packages)
 210         elif self.os in ["debian", "ubuntu"]:
 211             cmd = debfuncs.install_packages_command(self.os, packages)
 212         else:
 213             msg = "Error installing packages. OS not known for host %s " % (
 214                     self.get("hostname"))
 215             self.logger.error(msg)
 216             raise RuntimeError, msg
 217
 218         out = err = ""
 219         with self._lock:
 220             (out, err), proc = self.run_and_wait(cmd, self.home,
 221                 pidfile = "instpkgpid",
 222                 stdout = "instpkglog",
 223                 stderr = "instpkgerr",
 224                 raise_on_error = True)
 225
 226         return (out, err), proc
 227
 228     def remove_packages(self, packages):
 229         cmd = ""
 230         if self.os in ["f12", "f14"]:
 231             cmd = rpmfuncs.remove_packages_command(self.os, packages)
 232         elif self.os in ["debian", "ubuntu"]:
 233             cmd = debfuncs.remove_packages_command(self.os, packages)
 234         else:
 235             msg = "Error removing packages. OS not known for host %s " % (
 236                     self.get("hostname"))
 237             self.logger.error(msg)
 238             raise RuntimeError, msg
 239
 240         out = err = ""
 241         with self._lock:
 242             (out, err), proc = self.run_and_wait(cmd, self.home,
 243                 pidfile = "rmpkgpid",
 244                 stdout = "rmpkglog",
 245                 stderr = "rmpkgerr",
 246                 raise_on_error = True)
 247
 248         return (out, err), proc
 249
 250     def mkdir(self, path, clean = False):
 251         if clean:
 252             self.rmdir(path)
 253
 254         return self.execute("mkdir -p %s" % path)
 255
 256     def rmdir(self, path):
 257         return self.execute("rm -rf %s" % path)
 258
 259     def run_and_wait(self, command,
 260             home = ".",
 261             pidfile = "pid",
 262             stdin = None,
 263             stdout = 'stdout',
 264             stderr = 'stderr',
 265             sudo = False,
 266             raise_on_error = False):
 267
 268         (out, err), proc = self.run(command, home,
 269                 pidfile = pidfile,
 270                 stdin = stdin,
 271                 stdout = stdout,
 272                 stderr = stderr,
 273                 sudo = sudo)
 274
 275         if proc.poll() and err:
 276             msg = " Failed to run command %s on host %s" % (
 277                     command, self.get("hostname"))
 278             self.logger.error(msg)
 279             if raise_on_error:
 280                 raise RuntimeError, msg
 281
 282         pid, ppid = self.wait_pid(
 283                 home = home,
 284                 pidfile = pidfile,
 285                 raise_on_error = raise_on_error)
 286
 287         self.wait_run(pid, ppid)
 288
 289         (out, err), proc = self.check_run_error(home, stderr)
 290
 291         if err or out:
 292             msg = "Error while running command %s on host %s. error output: %s" % (
 293                     command, self.get("hostname"), out)
 294             if err:
 295                 msg += " . err: %s" % err
 296
 297             self.logger.error(msg)
 298             if raise_on_error:
 299                 raise RuntimeError, msg
 300
 301         return (out, err), proc
 302
 303     def wait_pid(self, home = ".", pidfile = "pid", raise_on_error = False):
 304         pid = ppid = None
 305         delay = 1.0
 306         for i in xrange(5):
 307             pidtuple = self.checkpid(home = home, pidfile = pidfile)
 308
 309             if pidtuple:
 310                 pid, ppid = pidtuple
 311                 break
 312             else:
 313                 time.sleep(delay)
 314                 delay = min(30,delay*1.2)
 315         else:
 316             msg = " Failed to get pid for pidfile %s/%s on host %s" % (
 317                     home, pidfile, self.get("hostname"))
 318             self.logger.error(msg)
 319             if raise_on_error:
 320                 raise RuntimeError, msg
 321
 322         return pid, ppid
 323
 324     def wait_run(self, pid, ppid, trial = 0):
 325         delay = 1.0
 326         first = True
 327         bustspin = 0
 328
 329         while True:
 330             status = self.status(pid, ppid)
 331
 332             if status is sshfuncs.FINISHED:
 333                 break
 334             elif status is not sshfuncs.RUNNING:
 335                 bustspin += 1
 336                 time.sleep(delay*(5.5+random.random()))
 337                 if bustspin > 12:
 338                     break
 339             else:
 340                 if first:
 341                     first = False
 342
 343                 time.sleep(delay*(0.5+random.random()))
 344                 delay = min(30,delay*1.2)
 345                 bustspin = 0
 346
 347     def check_run_error(self, home, stderr = 'stderr'):
 348         (out, err), proc = self.execute("cat %s" %
 349                 os.path.join(home, stderr))
 350         return (out, err), proc
 351
 352     def check_run_output(self, home, stdout = 'stdout'):
 353         (out, err), proc = self.execute("cat %s" %
 354                 os.path.join(home, stdout))
 355         return (out, err), proc
 356
 357
 358     def is_alive(self):
 359         if self.localhost:
 360             return True
 361
 362         out = err = ""
 363         try:
 364             (out, err), proc = self.execute("echo 'ALIVE'")
 365         except:
 366             import traceback
 367             trace = traceback.format_exc()
 368             self.logger.warn("Unresponsive host %s. got:\n out: %s err: %s\n traceback: %s",
 369                     self.get("hostname"), out, err, trace)
 370             return False
 371
 372         if out.strip().startswith('ALIVE'):
 373             return True
 374         else:
 375             self.logger.warn("Unresponsive host %s. got:\n%s%s",
 376                     self.get("hostname"), out, err)
 377             return False
 378
 379             # TODO!
 380             #if self.check_bad_host(out,err):
 381             #    self.blacklist()
 382
 383     def copy(self, src, dst):
 384         if self.localhost:
 385             (out, err), proc =  execfuncs.lcopy(source, dest,
 386                     recursive = True)
 387         else:
 388             (out, err), proc = self.safe_retry(sshfuncs.rcopy)(
 389                 src, dst,
 390                 port = self.get("port"),
 391                 identity = self.get("identity"),
 392                 server_key = self.get("serverKey"),
 393                 recursive = True)
 394
 395         return (out, err), proc
 396
 397     def execute(self, command,
 398             sudo = False,
 399             stdin = None,
 400             env = None,
 401             tty = False,
 402             forward_x11 = False,
 403             timeout = None,
 404             retry = 0,
 405             err_on_timeout = True,
 406             connect_timeout = 30,
 407             persistent = True
 408             ):
 409         """ Notice that this invocation will block until the
 410         execution finishes. If this is not the desired behavior,
 411         use 'run' instead."""
 412
 413         if self.localhost:
 414             (out, err), proc = execfuncs.lexec(command,
 415                     user = user,
 416                     sudo = sudo,
 417                     stdin = stdin,
 418                     env = env)
 419         else:
 420             (out, err), proc = self.safe_retry(sshfuncs.rexec)(
 421                     command,
 422                     host = self.get("hostname"),
 423                     user = self.get("username"),
 424                     port = self.get("port"),
 425                     agent = True,
 426                     sudo = sudo,
 427                     stdin = stdin,
 428                     identity = self.get("identity"),
 429                     server_key = self.get("serverKey"),
 430                     env = env,
 431                     tty = tty,
 432                     forward_x11 = forward_x11,
 433                     timeout = timeout,
 434                     retry = retry,
 435                     err_on_timeout = err_on_timeout,
 436                     connect_timeout = connect_timeout,
 437                     persistent = persistent
 438                     )
 439
 440         return (out, err), proc
 441
 442     def run(self, command,
 443             home = None,
 444             create_home = True,
 445             pidfile = "pid",
 446             stdin = None,
 447             stdout = 'stdout',
 448             stderr = 'stderr',
 449             sudo = False):
 450
 451         self.logger.info("Running %s", command)
 452
 453         if self.localhost:
 454             (out, err), proc = execfuncs.lspawn(command, pidfile,
 455                     stdout = stdout,
 456                     stderr = stderr,
 457                     stdin = stdin,
 458                     home = home,
 459                     create_home = create_home,
 460                     sudo = sudo,
 461                     user = user)
 462         else:
 463             # Start process in a "daemonized" way, using nohup and heavy
 464             # stdin/out redirection to avoid connection issues
 465             (out,err), proc = self.safe_retry(sshfuncs.rspawn)(
 466                 command,
 467                 pidfile = pidfile,
 468                 home = home,
 469                 create_home = create_home,
 470                 stdin = stdin if stdin is not None else '/dev/null',
 471                 stdout = stdout if stdout else '/dev/null',
 472                 stderr = stderr if stderr else '/dev/null',
 473                 sudo = sudo,
 474                 host = self.get("hostname"),
 475                 user = self.get("username"),
 476                 port = self.get("port"),
 477                 agent = True,
 478                 identity = self.get("identity"),
 479                 server_key = self.get("serverKey")
 480                 )
 481
 482         return (out, err), proc
 483
 484     def checkpid(self, home = ".", pidfile = "pid"):
 485         if self.localhost:
 486             pidtuple =  execfuncs.lcheckpid(os.path.join(home, pidfile))
 487         else:
 488             pidtuple = sshfuncs.rcheckpid(
 489                 os.path.join(home, pidfile),
 490                 host = self.get("hostname"),
 491                 user = self.get("username"),
 492                 port = self.get("port"),
 493                 agent = True,
 494                 identity = self.get("identity"),
 495                 server_key = self.get("serverKey")
 496                 )
 497
 498         return pidtuple
 499
 500     def status(self, pid, ppid):
 501         if self.localhost:
 502             status = execfuncs.lstatus(pid, ppid)
 503         else:
 504             status = sshfuncs.rstatus(
 505                     pid, ppid,
 506                     host = self.get("hostname"),
 507                     user = self.get("username"),
 508                     port = self.get("port"),
 509                     agent = True,
 510                     identity = self.get("identity"),
 511                     server_key = self.get("serverKey")
 512                     )
 513
 514         return status
 515
 516     def kill(self, pid, ppid, sudo = False):
 517         out = err = ""
 518         proc = None
 519         status = self.status(pid, ppid)
 520
 521         if status == sshfuncs.RUNNING:
 522             if self.localhost:
 523                 (out, err), proc = execfuncs.lkill(pid, ppid, sudo)
 524             else:
 525                 (out, err), proc = self.safe_retry(sshfuncs.rkill)(
 526                     pid, ppid,
 527                     host = self.get("hostname"),
 528                     user = self.get("username"),
 529                     port = self.get("port"),
 530                     agent = True,
 531                     sudo = sudo,
 532                     identity = self.get("identity"),
 533                     server_key = self.get("serverKey")
 534                     )
 535         return (out, err), proc
 536
 537     def check_bad_host(self, out, err):
 538         badre = re.compile(r'(?:'
 539                            r'|Error: disk I/O error'
 540                            r')',
 541                            re.I)
 542         return badre.search(out) or badre.search(err)
 543
 544     def blacklist(self):
 545         # TODO!!!!
 546         self.logger.warn("Blacklisting malfunctioning node %s", self.hostname)
 547         #import util
 548         #util.appendBlacklist(self.hostname)
 549
 550     def safe_retry(self, func):
 551         """Retries a function invocation using a lock"""
 552         import functools
 553         @functools.wraps(func)
 554         def rv(*p, **kw):
 555             fail_msg = " Failed to execute function %s(%s, %s) at host %s" % (
 556                 func.__name__, p, kw, self.get("hostname"))
 557             retry = kw.pop("_retry", False)
 558             wlock = kw.pop("_with_lock", False)
 559
 560             out = err = ""
 561             proc = None
 562             for i in xrange(0 if retry else 4):
 563                 try:
 564                     if wlock:
 565                         with self._lock:
 566                             (out, err), proc = func(*p, **kw)
 567                     else:
 568                         (out, err), proc = func(*p, **kw)
 569
 570                     if proc.poll():
 571                         if retry:
 572                             time.sleep(i*15)
 573                             continue
 574                         else:
 575                             self.logger.error("%s. out: %s error: %s", fail_msg, out, err)
 576                     break
 577                 except RuntimeError, e:
 578                     if x >= 3:
 579                         self.logger.error("%s. error: %s", fail_msg, e.args)
 580             return (out, err), proc
 581
 582         return rv
 583