Source Code for Module ganeti.server.noded

   1  # 
   2  # 
   3   
   4  # Copyright (C) 2006, 2007, 2010, 2011, 2012 Google Inc. 
   5  # 
   6  # This program is free software; you can redistribute it and/or modify 
   7  # it under the terms of the GNU General Public License as published by 
   8  # the Free Software Foundation; either version 2 of the License, or 
   9  # (at your option) any later version. 
  10  # 
  11  # This program is distributed in the hope that it will be useful, but 
  12  # WITHOUT ANY WARRANTY; without even the implied warranty of 
  13  # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU 
  14  # General Public License for more details. 
  15  # 
  16  # You should have received a copy of the GNU General Public License 
  17  # along with this program; if not, write to the Free Software 
  18  # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 
  19  # 02110-1301, USA. 
  20   
  21   
  22  """Ganeti node daemon""" 
  23   
  24  # pylint: disable=C0103,W0142 
  25   
  26  # C0103: Functions in this module need to have a given name structure, 
  27  # and the name of the daemon doesn't match 
  28   
  29  # W0142: Used * or ** magic, since we do use it extensively in this 
  30  # module 
  31   
  32  import os 
  33  import sys 
  34  import logging 
  35  import signal 
  36  import codecs 
  37   
  38  from optparse import OptionParser 
  39   
  40  from ganeti import backend 
  41  from ganeti import constants 
  42  from ganeti import objects 
  43  from ganeti import errors 
  44  from ganeti import jstore 
  45  from ganeti import daemon 
  46  from ganeti import http 
  47  from ganeti import utils 
  48  from ganeti import storage 
  49  from ganeti import serializer 
  50  from ganeti import netutils 
  51  from ganeti import pathutils 
  52  from ganeti import ssconf 
  53   
  54  import ganeti.http.server # pylint: disable=W0611 
  55   
  56   
  57  queue_lock = None 
  58   
  59   
  60 -def _extendReasonTrail(trail, source, reason=""): 
  61    """Extend the reason trail with noded information 
  62   
  63    The trail is extended by appending the name of the noded functionality 
  64    """ 
  65    assert trail is not None 
  66    trail_source = "%s:%s" % (constants.OPCODE_REASON_SRC_NODED, source) 
  67    trail.append((trail_source, reason, utils.EpochNano())) 
  68   
  69   
  70 -def _PrepareQueueLock(): 
  71    """Try to prepare the queue lock. 
  72   
  73    @return: None for success, otherwise an exception object 
  74   
  75    """ 
  76    global queue_lock # pylint: disable=W0603 
  77   
  78    if queue_lock is not None: 
  79      return None 
  80   
  81    # Prepare job queue 
  82    try: 
  83      queue_lock = jstore.InitAndVerifyQueue(must_lock=False) 
  84      return None 
  85    except EnvironmentError, err: 
  86      return err 
  87   
  88   
  89 -def _RequireJobQueueLock(fn): 
  90    """Decorator for job queue manipulating functions. 
  91   
  92    """ 
  93    QUEUE_LOCK_TIMEOUT = 10 
  94   
  95    def wrapper(*args, **kwargs): 
  96      # Locking in exclusive, blocking mode because there could be several 
  97      # children running at the same time. Waiting up to 10 seconds. 
  98      if _PrepareQueueLock() is not None: 
  99        raise errors.JobQueueError("Job queue failed initialization," 
 100                                   " cannot update jobs") 
 101      queue_lock.Exclusive(blocking=True, timeout=QUEUE_LOCK_TIMEOUT) 
 102      try: 
 103        return fn(*args, **kwargs) 
 104      finally: 
 105        queue_lock.Unlock() 
 106   
 107    return wrapper 
 108   
 109   
 110 -def _DecodeImportExportIO(ieio, ieioargs): 
 111    """Decodes import/export I/O information. 
 112   
 113    """ 
 114    if ieio == constants.IEIO_RAW_DISK: 
 115      assert len(ieioargs) == 1 
 116      return (objects.Disk.FromDict(ieioargs[0]), ) 
 117   
 118    if ieio == constants.IEIO_SCRIPT: 
 119      assert len(ieioargs) == 2 
 120      return (objects.Disk.FromDict(ieioargs[0]), ieioargs[1]) 
 121   
 122    return ieioargs 
 123   
 124   
 125 -def _DefaultAlternative(value, default): 
 126    """Returns value or, if evaluating to False, a default value. 
 127   
 128    Returns the given value, unless it evaluates to False. In the latter case the 
 129    default value is returned. 
 130   
 131    @param value: Value to return if it doesn't evaluate to False 
 132    @param default: Default value 
 133    @return: Given value or the default 
 134   
 135    """ 
 136    if value: 
 137      return value 
 138   
 139    return default 
 140   
 141   
 142 -class MlockallRequestExecutor(http.server.HttpServerRequestExecutor): 
 143    """Subclass ensuring request handlers are locked in RAM. 
 144   
 145    """ 
 146 -  def __init__(self, *args, **kwargs): 
 147      utils.Mlockall() 
 148   
 149      http.server.HttpServerRequestExecutor.__init__(self, *args, **kwargs) 
 150   
 151   
 152 -class NodeRequestHandler(http.server.HttpServerHandler): 
 153    """The server implementation. 
 154   
 155    This class holds all methods exposed over the RPC interface. 
 156   
 157    """ 
 158    # too many public methods, and unused args - all methods get params 
 159    # due to the API 
 160    # pylint: disable=R0904,W0613 
 161 -  def __init__(self): 
 162      http.server.HttpServerHandler.__init__(self) 
 163      self.noded_pid = os.getpid() 
 164   
 165 -  def HandleRequest(self, req): 
 166      """Handle a request. 
 167   
 168      """ 
 169      if req.request_method.upper() != http.HTTP_POST: 
 170        raise http.HttpBadRequest("Only the POST method is supported") 
 171   
 172      path = req.request_path 
 173      if path.startswith("/"): 
 174        path = path[1:] 
 175   
 176      method = getattr(self, "perspective_%s" % path, None) 
 177      if method is None: 
 178        raise http.HttpNotFound() 
 179   
 180      try: 
 181        result = (True, method(serializer.LoadJson(req.request_body))) 
 182   
 183      except backend.RPCFail, err: 
 184        # our custom failure exception; str(err) works fine if the 
 185        # exception was constructed with a single argument, and in 
 186        # this case, err.message == err.args[0] == str(err) 
 187        result = (False, str(err)) 
 188      except errors.QuitGanetiException, err: 
 189        # Tell parent to quit 
 190        logging.info("Shutting down the node daemon, arguments: %s", 
 191                     str(err.args)) 
 192        os.kill(self.noded_pid, signal.SIGTERM) 
 193        # And return the error's arguments, which must be already in 
 194        # correct tuple format 
 195        result = err.args 
 196      except Exception, err: 
 197        logging.exception("Error in RPC call") 
 198        result = (False, "Error while executing backend function: %s" % str(err)) 
 199   
 200      return serializer.DumpJson(result) 
 201   
 202    # the new block devices  -------------------------- 
 203   
 204    @staticmethod 
 205 -  def perspective_blockdev_create(params): 
 206      """Create a block device. 
 207   
 208      """ 
 209      (bdev_s, size, owner, on_primary, info, excl_stor) = params 
 210      bdev = objects.Disk.FromDict(bdev_s) 
 211      if bdev is None: 
 212        raise ValueError("can't unserialize data!") 
 213      return backend.BlockdevCreate(bdev, size, owner, on_primary, info, 
 214                                    excl_stor) 
 215   
 216    @staticmethod 
 217 -  def perspective_blockdev_pause_resume_sync(params): 
 218      """Pause/resume sync of a block device. 
 219   
 220      """ 
 221      disks_s, pause = params 
 222      disks = [objects.Disk.FromDict(bdev_s) for bdev_s in disks_s] 
 223      return backend.BlockdevPauseResumeSync(disks, pause) 
 224   
 225    @staticmethod 
 226 -  def perspective_blockdev_wipe(params): 
 227      """Wipe a block device. 
 228   
 229      """ 
 230      bdev_s, offset, size = params 
 231      bdev = objects.Disk.FromDict(bdev_s) 
 232      return backend.BlockdevWipe(bdev, offset, size) 
 233   
 234    @staticmethod 
 235 -  def perspective_blockdev_remove(params): 
 236      """Remove a block device. 
 237   
 238      """ 
 239      bdev_s = params[0] 
 240      bdev = objects.Disk.FromDict(bdev_s) 
 241      return backend.BlockdevRemove(bdev) 
 242   
 243    @staticmethod 
 244 -  def perspective_blockdev_rename(params): 
 245      """Remove a block device. 
 246   
 247      """ 
 248      devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params[0]] 
 249      return backend.BlockdevRename(devlist) 
 250   
 251    @staticmethod 
 252 -  def perspective_blockdev_assemble(params): 
 253      """Assemble a block device. 
 254   
 255      """ 
 256      bdev_s, owner, on_primary, idx = params 
 257      bdev = objects.Disk.FromDict(bdev_s) 
 258      if bdev is None: 
 259        raise ValueError("can't unserialize data!") 
 260      return backend.BlockdevAssemble(bdev, owner, on_primary, idx) 
 261   
 262    @staticmethod 
 263 -  def perspective_blockdev_shutdown(params): 
 264      """Shutdown a block device. 
 265   
 266      """ 
 267      bdev_s = params[0] 
 268      bdev = objects.Disk.FromDict(bdev_s) 
 269      if bdev is None: 
 270        raise ValueError("can't unserialize data!") 
 271      return backend.BlockdevShutdown(bdev) 
 272   
 273    @staticmethod 
 274 -  def perspective_blockdev_addchildren(params): 
 275      """Add a child to a mirror device. 
 276   
 277      Note: this is only valid for mirror devices. It's the caller's duty 
 278      to send a correct disk, otherwise we raise an error. 
 279   
 280      """ 
 281      bdev_s, ndev_s = params 
 282      bdev = objects.Disk.FromDict(bdev_s) 
 283      ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s] 
 284      if bdev is None or ndevs.count(None) > 0: 
 285        raise ValueError("can't unserialize data!") 
 286      return backend.BlockdevAddchildren(bdev, ndevs) 
 287   
 288    @staticmethod 
 289 -  def perspective_blockdev_removechildren(params): 
 290      """Remove a child from a mirror device. 
 291   
 292      This is only valid for mirror devices, of course. It's the callers 
 293      duty to send a correct disk, otherwise we raise an error. 
 294   
 295      """ 
 296      bdev_s, ndev_s = params 
 297      bdev = objects.Disk.FromDict(bdev_s) 
 298      ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s] 
 299      if bdev is None or ndevs.count(None) > 0: 
 300        raise ValueError("can't unserialize data!") 
 301      return backend.BlockdevRemovechildren(bdev, ndevs) 
 302   
 303    @staticmethod 
 304 -  def perspective_blockdev_getmirrorstatus(params): 
 305      """Return the mirror status for a list of disks. 
 306   
 307      """ 
 308      disks = [objects.Disk.FromDict(dsk_s) 
 309               for dsk_s in params[0]] 
 310      return [status.ToDict() 
 311              for status in backend.BlockdevGetmirrorstatus(disks)] 
 312   
 313    @staticmethod 
 314 -  def perspective_blockdev_getmirrorstatus_multi(params): 
 315      """Return the mirror status for a list of disks. 
 316   
 317      """ 
 318      (node_disks, ) = params 
 319   
 320      disks = [objects.Disk.FromDict(dsk_s) for dsk_s in node_disks] 
 321   
 322      result = [] 
 323   
 324      for (success, status) in backend.BlockdevGetmirrorstatusMulti(disks): 
 325        if success: 
 326          result.append((success, status.ToDict())) 
 327        else: 
 328          result.append((success, status)) 
 329   
 330      return result 
 331   
 332    @staticmethod 
 333 -  def perspective_blockdev_find(params): 
 334      """Expose the FindBlockDevice functionality for a disk. 
 335   
 336      This will try to find but not activate a disk. 
 337   
 338      """ 
 339      disk = objects.Disk.FromDict(params[0]) 
 340   
 341      result = backend.BlockdevFind(disk) 
 342      if result is None: 
 343        return None 
 344   
 345      return result.ToDict() 
 346   
 347    @staticmethod 
 348 -  def perspective_blockdev_snapshot(params): 
 349      """Create a snapshot device. 
 350   
 351      Note that this is only valid for LVM disks, if we get passed 
 352      something else we raise an exception. The snapshot device can be 
 353      remove by calling the generic block device remove call. 
 354   
 355      """ 
 356      cfbd = objects.Disk.FromDict(params[0]) 
 357      return backend.BlockdevSnapshot(cfbd) 
 358   
 359    @staticmethod 
 360 -  def perspective_blockdev_grow(params): 
 361      """Grow a stack of devices. 
 362   
 363      """ 
 364      if len(params) < 4: 
 365        raise ValueError("Received only 3 parameters in blockdev_grow," 
 366                         " old master?") 
 367      cfbd = objects.Disk.FromDict(params[0]) 
 368      amount = params[1] 
 369      dryrun = params[2] 
 370      backingstore = params[3] 
 371      return backend.BlockdevGrow(cfbd, amount, dryrun, backingstore) 
 372   
 373    @staticmethod 
 374 -  def perspective_blockdev_close(params): 
 375      """Closes the given block devices. 
 376   
 377      """ 
 378      disks = [objects.Disk.FromDict(cf) for cf in params[1]] 
 379      return backend.BlockdevClose(params[0], disks) 
 380   
 381    @staticmethod 
 382 -  def perspective_blockdev_getsize(params): 
 383      """Compute the sizes of the given block devices. 
 384   
 385      """ 
 386      disks = [objects.Disk.FromDict(cf) for cf in params[0]] 
 387      return backend.BlockdevGetsize(disks) 
 388   
 389    @staticmethod 
 390 -  def perspective_blockdev_export(params): 
 391      """Compute the sizes of the given block devices. 
 392   
 393      """ 
 394      disk = objects.Disk.FromDict(params[0]) 
 395      dest_node, dest_path, cluster_name = params[1:] 
 396      return backend.BlockdevExport(disk, dest_node, dest_path, cluster_name) 
 397   
 398    @staticmethod 
 399 -  def perspective_blockdev_setinfo(params): 
 400      """Sets metadata information on the given block device. 
 401   
 402      """ 
 403      (disk, info) = params 
 404      disk = objects.Disk.FromDict(disk) 
 405      return backend.BlockdevSetInfo(disk, info) 
 406   
 407    # blockdev/drbd specific methods ---------- 
 408   
 409    @staticmethod 
 410 -  def perspective_drbd_disconnect_net(params): 
 411      """Disconnects the network connection of drbd disks. 
 412   
 413      Note that this is only valid for drbd disks, so the members of the 
 414      disk list must all be drbd devices. 
 415   
 416      """ 
 417      nodes_ip, disks = params 
 418      disks = [objects.Disk.FromDict(cf) for cf in disks] 
 419      return backend.DrbdDisconnectNet(nodes_ip, disks) 
 420   
 421    @staticmethod 
 422 -  def perspective_drbd_attach_net(params): 
 423      """Attaches the network connection of drbd disks. 
 424   
 425      Note that this is only valid for drbd disks, so the members of the 
 426      disk list must all be drbd devices. 
 427   
 428      """ 
 429      nodes_ip, disks, instance_name, multimaster = params 
 430      disks = [objects.Disk.FromDict(cf) for cf in disks] 
 431      return backend.DrbdAttachNet(nodes_ip, disks, 
 432                                       instance_name, multimaster) 
 433   
 434    @staticmethod 
 435 -  def perspective_drbd_wait_sync(params): 
 436      """Wait until DRBD disks are synched. 
 437   
 438      Note that this is only valid for drbd disks, so the members of the 
 439      disk list must all be drbd devices. 
 440   
 441      """ 
 442      nodes_ip, disks = params 
 443      disks = [objects.Disk.FromDict(cf) for cf in disks] 
 444      return backend.DrbdWaitSync(nodes_ip, disks) 
 445   
 446    @staticmethod 
 447 -  def perspective_drbd_helper(params): 
 448      """Query drbd helper. 
 449   
 450      """ 
 451      return backend.GetDrbdUsermodeHelper() 
 452   
 453    # export/import  -------------------------- 
 454   
 455    @staticmethod 
 456 -  def perspective_finalize_export(params): 
 457      """Expose the finalize export functionality. 
 458   
 459      """ 
 460      instance = objects.Instance.FromDict(params[0]) 
 461   
 462      snap_disks = [] 
 463      for disk in params[1]: 
 464        if isinstance(disk, bool): 
 465          snap_disks.append(disk) 
 466        else: 
 467          snap_disks.append(objects.Disk.FromDict(disk)) 
 468   
 469      return backend.FinalizeExport(instance, snap_disks) 
 470   
 471    @staticmethod 
 472 -  def perspective_export_info(params): 
 473      """Query information about an existing export on this node. 
 474   
 475      The given path may not contain an export, in which case we return 
 476      None. 
 477   
 478      """ 
 479      path = params[0] 
 480      return backend.ExportInfo(path) 
 481   
 482    @staticmethod 
 483 -  def perspective_export_list(params): 
 484      """List the available exports on this node. 
 485   
 486      Note that as opposed to export_info, which may query data about an 
 487      export in any path, this only queries the standard Ganeti path 
 488      (pathutils.EXPORT_DIR). 
 489   
 490      """ 
 491      return backend.ListExports() 
 492   
 493    @staticmethod 
 494 -  def perspective_export_remove(params): 
 495      """Remove an export. 
 496   
 497      """ 
 498      export = params[0] 
 499      return backend.RemoveExport(export) 
 500   
 501    # block device --------------------- 
 502    @staticmethod 
 503 -  def perspective_bdev_sizes(params): 
 504      """Query the list of block devices 
 505   
 506      """ 
 507      devices = params[0] 
 508      return backend.GetBlockDevSizes(devices) 
 509   
 510    # volume  -------------------------- 
 511   
 512    @staticmethod 
 513 -  def perspective_lv_list(params): 
 514      """Query the list of logical volumes in a given volume group. 
 515   
 516      """ 
 517      vgname = params[0] 
 518      return backend.GetVolumeList(vgname) 
 519   
 520    @staticmethod 
 521 -  def perspective_vg_list(params): 
 522      """Query the list of volume groups. 
 523   
 524      """ 
 525      return backend.ListVolumeGroups() 
 526   
 527    # Storage -------------------------- 
 528   
 529    @staticmethod 
 530 -  def perspective_storage_list(params): 
 531      """Get list of storage units. 
 532   
 533      """ 
 534      (su_name, su_args, name, fields) = params 
 535      return storage.GetStorage(su_name, *su_args).List(name, fields) 
 536   
 537    @staticmethod 
 538 -  def perspective_storage_modify(params): 
 539      """Modify a storage unit. 
 540   
 541      """ 
 542      (su_name, su_args, name, changes) = params 
 543      return storage.GetStorage(su_name, *su_args).Modify(name, changes) 
 544   
 545    @staticmethod 
 546 -  def perspective_storage_execute(params): 
 547      """Execute an operation on a storage unit. 
 548   
 549      """ 
 550      (su_name, su_args, name, op) = params 
 551      return storage.GetStorage(su_name, *su_args).Execute(name, op) 
 552   
 553    # bridge  -------------------------- 
 554   
 555    @staticmethod 
 556 -  def perspective_bridges_exist(params): 
 557      """Check if all bridges given exist on this node. 
 558   
 559      """ 
 560      bridges_list = params[0] 
 561      return backend.BridgesExist(bridges_list) 
 562   
 563    # instance  -------------------------- 
 564   
 565    @staticmethod 
 566 -  def perspective_instance_os_add(params): 
 567      """Install an OS on a given instance. 
 568   
 569      """ 
 570      inst_s = params[0] 
 571      inst = objects.Instance.FromDict(inst_s) 
 572      reinstall = params[1] 
 573      debug = params[2] 
 574      return backend.InstanceOsAdd(inst, reinstall, debug) 
 575   
 576    @staticmethod 
 577 -  def perspective_instance_run_rename(params): 
 578      """Runs the OS rename script for an instance. 
 579   
 580      """ 
 581      inst_s, old_name, debug = params 
 582      inst = objects.Instance.FromDict(inst_s) 
 583      return backend.RunRenameInstance(inst, old_name, debug) 
 584   
 585    @staticmethod 
 586 -  def perspective_instance_shutdown(params): 
 587      """Shutdown an instance. 
 588   
 589      """ 
 590      instance = objects.Instance.FromDict(params[0]) 
 591      timeout = params[1] 
 592      trail = params[2] 
 593      _extendReasonTrail(trail, "shutdown") 
 594      return backend.InstanceShutdown(instance, timeout, trail) 
 595   
 596    @staticmethod 
 597 -  def perspective_instance_start(params): 
 598      """Start an instance. 
 599   
 600      """ 
 601      (instance_name, startup_paused, trail) = params 
 602      instance = objects.Instance.FromDict(instance_name) 
 603      _extendReasonTrail(trail, "start") 
 604      return backend.StartInstance(instance, startup_paused, trail) 
 605   
 606    @staticmethod 
 607 -  def perspective_migration_info(params): 
 608      """Gather information about an instance to be migrated. 
 609   
 610      """ 
 611      instance = objects.Instance.FromDict(params[0]) 
 612      return backend.MigrationInfo(instance) 
 613   
 614    @staticmethod 
 615 -  def perspective_accept_instance(params): 
 616      """Prepare the node to accept an instance. 
 617   
 618      """ 
 619      instance, info, target = params 
 620      instance = objects.Instance.FromDict(instance) 
 621      return backend.AcceptInstance(instance, info, target) 
 622   
 623    @staticmethod 
 624 -  def perspective_instance_finalize_migration_dst(params): 
 625      """Finalize the instance migration on the destination node. 
 626   
 627      """ 
 628      instance, info, success = params 
 629      instance = objects.Instance.FromDict(instance) 
 630      return backend.FinalizeMigrationDst(instance, info, success) 
 631   
 632    @staticmethod 
 633 -  def perspective_instance_migrate(params): 
 634      """Migrates an instance. 
 635   
 636      """ 
 637      instance, target, live = params 
 638      instance = objects.Instance.FromDict(instance) 
 639      return backend.MigrateInstance(instance, target, live) 
 640   
 641    @staticmethod 
 642 -  def perspective_instance_finalize_migration_src(params): 
 643      """Finalize the instance migration on the source node. 
 644   
 645      """ 
 646      instance, success, live = params 
 647      instance = objects.Instance.FromDict(instance) 
 648      return backend.FinalizeMigrationSource(instance, success, live) 
 649   
 650    @staticmethod 
 651 -  def perspective_instance_get_migration_status(params): 
 652      """Reports migration status. 
 653   
 654      """ 
 655      instance = objects.Instance.FromDict(params[0]) 
 656      return backend.GetMigrationStatus(instance).ToDict() 
 657   
 658    @staticmethod 
 659 -  def perspective_instance_reboot(params): 
 660      """Reboot an instance. 
 661   
 662      """ 
 663      instance = objects.Instance.FromDict(params[0]) 
 664      reboot_type = params[1] 
 665      shutdown_timeout = params[2] 
 666      trail = params[3] 
 667      _extendReasonTrail(trail, "reboot") 
 668      return backend.InstanceReboot(instance, reboot_type, shutdown_timeout, 
 669                                    trail) 
 670   
 671    @staticmethod 
 672 -  def perspective_instance_balloon_memory(params): 
 673      """Modify instance runtime memory. 
 674   
 675      """ 
 676      instance_dict, memory = params 
 677      instance = objects.Instance.FromDict(instance_dict) 
 678      return backend.InstanceBalloonMemory(instance, memory) 
 679   
 680    @staticmethod 
 681 -  def perspective_instance_info(params): 
 682      """Query instance information. 
 683   
 684      """ 
 685      return backend.GetInstanceInfo(params[0], params[1]) 
 686   
 687    @staticmethod 
 688 -  def perspective_instance_migratable(params): 
 689      """Query whether the specified instance can be migrated. 
 690   
 691      """ 
 692      instance = objects.Instance.FromDict(params[0]) 
 693      return backend.GetInstanceMigratable(instance) 
 694   
 695    @staticmethod 
 696 -  def perspective_all_instances_info(params): 
 697      """Query information about all instances. 
 698   
 699      """ 
 700      return backend.GetAllInstancesInfo(params[0]) 
 701   
 702    @staticmethod 
 703 -  def perspective_instance_list(params): 
 704      """Query the list of running instances. 
 705   
 706      """ 
 707      return backend.GetInstanceList(params[0]) 
 708   
 709    # node -------------------------- 
 710   
 711    @staticmethod 
 712 -  def perspective_node_has_ip_address(params): 
 713      """Checks if a node has the given ip address. 
 714   
 715      """ 
 716      return netutils.IPAddress.Own(params[0]) 
 717   
 718    @staticmethod 
 719 -  def perspective_node_info(params): 
 720      """Query node information. 
 721   
 722      """ 
 723      (vg_names, hv_names, excl_stor) = params 
 724      return backend.GetNodeInfo(vg_names, hv_names, excl_stor) 
 725   
 726    @staticmethod 
 727 -  def perspective_etc_hosts_modify(params): 
 728      """Modify a node entry in /etc/hosts. 
 729   
 730      """ 
 731      backend.EtcHostsModify(params[0], params[1], params[2]) 
 732   
 733      return True 
 734   
 735    @staticmethod 
 736 -  def perspective_node_verify(params): 
 737      """Run a verify sequence on this node. 
 738   
 739      """ 
 740      return backend.VerifyNode(params[0], params[1]) 
 741   
 742    @classmethod 
 743 -  def perspective_node_verify_light(cls, params): 
 744      """Run a light verify sequence on this node. 
 745   
 746      """ 
 747      # So far it's the same as the normal node_verify 
 748      return cls.perspective_node_verify(params) 
 749   
 750    @staticmethod 
 751 -  def perspective_node_start_master_daemons(params): 
 752      """Start the master daemons on this node. 
 753   
 754      """ 
 755      return backend.StartMasterDaemons(params[0]) 
 756   
 757    @staticmethod 
 758 -  def perspective_node_activate_master_ip(params): 
 759      """Activate the master IP on this node. 
 760   
 761      """ 
 762      master_params = objects.MasterNetworkParameters.FromDict(params[0]) 
 763      return backend.ActivateMasterIp(master_params, params[1]) 
 764   
 765    @staticmethod 
 766 -  def perspective_node_deactivate_master_ip(params): 
 767      """Deactivate the master IP on this node. 
 768   
 769      """ 
 770      master_params = objects.MasterNetworkParameters.FromDict(params[0]) 
 771      return backend.DeactivateMasterIp(master_params, params[1]) 
 772   
 773    @staticmethod 
 774 -  def perspective_node_stop_master(params): 
 775      """Stops master daemons on this node. 
 776   
 777      """ 
 778      return backend.StopMasterDaemons() 
 779   
 780    @staticmethod 
 781 -  def perspective_node_change_master_netmask(params): 
 782      """Change the master IP netmask. 
 783   
 784      """ 
 785      return backend.ChangeMasterNetmask(params[0], params[1], params[2], 
 786                                         params[3]) 
 787   
 788    @staticmethod 
 789 -  def perspective_node_leave_cluster(params): 
 790      """Cleanup after leaving a cluster. 
 791   
 792      """ 
 793      return backend.LeaveCluster(params[0]) 
 794   
 795    @staticmethod 
 796 -  def perspective_node_volumes(params): 
 797      """Query the list of all logical volume groups. 
 798   
 799      """ 
 800      return backend.NodeVolumes() 
 801   
 802    @staticmethod 
 803 -  def perspective_node_demote_from_mc(params): 
 804      """Demote a node from the master candidate role. 
 805   
 806      """ 
 807      return backend.DemoteFromMC() 
 808   
 809    @staticmethod 
 810 -  def perspective_node_powercycle(params): 
 811      """Tries to powercycle the nod. 
 812   
 813      """ 
 814      hypervisor_type = params[0] 
 815      return backend.PowercycleNode(hypervisor_type) 
 816   
 817    # cluster -------------------------- 
 818   
 819    @staticmethod 
 820 -  def perspective_version(params): 
 821      """Query version information. 
 822   
 823      """ 
 824      return constants.PROTOCOL_VERSION 
 825   
 826    @staticmethod 
 827 -  def perspective_upload_file(params): 
 828      """Upload a file. 
 829   
 830      Note that the backend implementation imposes strict rules on which 
 831      files are accepted. 
 832   
 833      """ 
 834      return backend.UploadFile(*(params[0])) 
 835   
 836    @staticmethod 
 837 -  def perspective_master_info(params): 
 838      """Query master information. 
 839   
 840      """ 
 841      return backend.GetMasterInfo() 
 842   
 843    @staticmethod 
 844 -  def perspective_run_oob(params): 
 845      """Runs oob on node. 
 846   
 847      """ 
 848      output = backend.RunOob(params[0], params[1], params[2], params[3]) 
 849      if output: 
 850        result = serializer.LoadJson(output) 
 851      else: 
 852        result = None 
 853      return result 
 854   
 855    @staticmethod 
 856 -  def perspective_restricted_command(params): 
 857      """Runs a restricted command. 
 858   
 859      """ 
 860      (cmd, ) = params 
 861   
 862      return backend.RunRestrictedCmd(cmd) 
 863   
 864    @staticmethod 
 865 -  def perspective_write_ssconf_files(params): 
 866      """Write ssconf files. 
 867   
 868      """ 
 869      (values,) = params 
 870      return ssconf.WriteSsconfFiles(values) 
 871   
 872    @staticmethod 
 873 -  def perspective_get_watcher_pause(params): 
 874      """Get watcher pause end. 
 875   
 876      """ 
 877      return utils.ReadWatcherPauseFile(pathutils.WATCHER_PAUSEFILE) 
 878   
 879    @staticmethod 
 880 -  def perspective_set_watcher_pause(params): 
 881      """Set watcher pause. 
 882   
 883      """ 
 884      (until, ) = params 
 885      return backend.SetWatcherPause(until) 
 886   
 887    # os ----------------------- 
 888   
 889    @staticmethod 
 890 -  def perspective_os_diagnose(params): 
 891      """Query detailed information about existing OSes. 
 892   
 893      """ 
 894      return backend.DiagnoseOS() 
 895   
 896    @staticmethod 
 897 -  def perspective_os_get(params): 
 898      """Query information about a given OS. 
 899   
 900      """ 
 901      name = params[0] 
 902      os_obj = backend.OSFromDisk(name) 
 903      return os_obj.ToDict() 
 904   
 905    @staticmethod 
 906 -  def perspective_os_validate(params): 
 907      """Run a given OS' validation routine. 
 908   
 909      """ 
 910      required, name, checks, params = params 
 911      return backend.ValidateOS(required, name, checks, params) 
 912   
 913    # extstorage ----------------------- 
 914   
 915    @staticmethod 
 916 -  def perspective_extstorage_diagnose(params): 
 917      """Query detailed information about existing extstorage providers. 
 918   
 919      """ 
 920      return backend.DiagnoseExtStorage() 
 921   
 922    # hooks ----------------------- 
 923   
 924    @staticmethod 
 925 -  def perspective_hooks_runner(params): 
 926      """Run hook scripts. 
 927   
 928      """ 
 929      hpath, phase, env = params 
 930      hr = backend.HooksRunner() 
 931      return hr.RunHooks(hpath, phase, env) 
 932   
 933    # iallocator ----------------- 
 934   
 935    @staticmethod 
 936 -  def perspective_iallocator_runner(params): 
 937      """Run an iallocator script. 
 938   
 939      """ 
 940      name, idata = params 
 941      iar = backend.IAllocatorRunner() 
 942      return iar.Run(name, idata) 
 943   
 944    # test ----------------------- 
 945   
 946    @staticmethod 
 947 -  def perspective_test_delay(params): 
 948      """Run test delay. 
 949   
 950      """ 
 951      duration = params[0] 
 952      status, rval = utils.TestDelay(duration) 
 953      if not status: 
 954        raise backend.RPCFail(rval) 
 955      return rval 
 956   
 957    # file storage --------------- 
 958   
 959    @staticmethod 
 960 -  def perspective_file_storage_dir_create(params): 
 961      """Create the file storage directory. 
 962   
 963      """ 
 964      file_storage_dir = params[0] 
 965      return backend.CreateFileStorageDir(file_storage_dir) 
 966   
 967    @staticmethod 
 968 -  def perspective_file_storage_dir_remove(params): 
 969      """Remove the file storage directory. 
 970   
 971      """ 
 972      file_storage_dir = params[0] 
 973      return backend.RemoveFileStorageDir(file_storage_dir) 
 974   
 975    @staticmethod 
 976 -  def perspective_file_storage_dir_rename(params): 
 977      """Rename the file storage directory. 
 978   
 979      """ 
 980      old_file_storage_dir = params[0] 
 981      new_file_storage_dir = params[1] 
 982      return backend.RenameFileStorageDir(old_file_storage_dir, 
 983                                          new_file_storage_dir) 
 984   
 985    # jobs ------------------------ 
 986   
 987    @staticmethod 
 988    @_RequireJobQueueLock 
 989 -  def perspective_jobqueue_update(params): 
 990      """Update job queue. 
 991   
 992      """ 
 993      (file_name, content) = params 
 994      return backend.JobQueueUpdate(file_name, content) 
 995   
 996    @staticmethod 
 997    @_RequireJobQueueLock 
 998 -  def perspective_jobqueue_purge(params): 
 999      """Purge job queue. 
1000   
1001      """ 
1002      return backend.JobQueuePurge() 
1003   
1004    @staticmethod 
1005    @_RequireJobQueueLock 
1006 -  def perspective_jobqueue_rename(params): 
1007      """Rename a job queue file. 
1008   
1009      """ 
1010      # TODO: What if a file fails to rename? 
1011      return [backend.JobQueueRename(old, new) for old, new in params[0]] 
1012   
1013    @staticmethod 
1014    @_RequireJobQueueLock 
1015 -  def perspective_jobqueue_set_drain_flag(params): 
1016      """Set job queue's drain flag. 
1017   
1018      """ 
1019      (flag, ) = params 
1020   
1021      return jstore.SetDrainFlag(flag) 
1022   
1023    # hypervisor --------------- 
1024   
1025    @staticmethod 
1026 -  def perspective_hypervisor_validate_params(params): 
1027      """Validate the hypervisor parameters. 
1028   
1029      """ 
1030      (hvname, hvparams) = params 
1031      return backend.ValidateHVParams(hvname, hvparams) 
1032   
1033    # Crypto 
1034   
1035    @staticmethod 
1036 -  def perspective_x509_cert_create(params): 
1037      """Creates a new X509 certificate for SSL/TLS. 
1038   
1039      """ 
1040      (validity, ) = params 
1041      return backend.CreateX509Certificate(validity) 
1042   
1043    @staticmethod 
1044 -  def perspective_x509_cert_remove(params): 
1045      """Removes a X509 certificate. 
1046   
1047      """ 
1048      (name, ) = params 
1049      return backend.RemoveX509Certificate(name) 
1050   
1051    # Import and export 
1052   
1053    @staticmethod 
1054 -  def perspective_import_start(params): 
1055      """Starts an import daemon. 
1056   
1057      """ 
1058      (opts_s, instance, component, (dest, dest_args)) = params 
1059   
1060      opts = objects.ImportExportOptions.FromDict(opts_s) 
1061   
1062      return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts, 
1063                                             None, None, 
1064                                             objects.Instance.FromDict(instance), 
1065                                             component, dest, 
1066                                             _DecodeImportExportIO(dest, 
1067                                                                   dest_args)) 
1068   
1069    @staticmethod 
1070 -  def perspective_export_start(params): 
1071      """Starts an export daemon. 
1072   
1073      """ 
1074      (opts_s, host, port, instance, component, (source, source_args)) = params 
1075   
1076      opts = objects.ImportExportOptions.FromDict(opts_s) 
1077   
1078      return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts, 
1079                                             host, port, 
1080                                             objects.Instance.FromDict(instance), 
1081                                             component, source, 
1082                                             _DecodeImportExportIO(source, 
1083                                                                   source_args)) 
1084   
1085    @staticmethod 
1086 -  def perspective_impexp_status(params): 
1087      """Retrieves the status of an import or export daemon. 
1088   
1089      """ 
1090      return backend.GetImportExportStatus(params[0]) 
1091   
1092    @staticmethod 
1093 -  def perspective_impexp_abort(params): 
1094      """Aborts an import or export. 
1095   
1096      """ 
1097      return backend.AbortImportExport(params[0]) 
1098   
1099    @staticmethod 
1100 -  def perspective_impexp_cleanup(params): 
1101      """Cleans up after an import or export. 
1102   
1103      """ 
1104      return backend.CleanupImportExport(params[0]) 
1105   
1106   
1107 -def CheckNoded(_, args): 
1108    """Initial checks whether to run or exit with a failure. 
1109   
1110    """ 
1111    if args: # noded doesn't take any arguments 
1112      print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" % 
1113                            sys.argv[0]) 
1114      sys.exit(constants.EXIT_FAILURE) 
1115    try: 
1116      codecs.lookup("string-escape") 
1117    except LookupError: 
1118      print >> sys.stderr, ("Can't load the string-escape code which is part" 
1119                            " of the Python installation. Is your installation" 
1120                            " complete/correct? Aborting.") 
1121      sys.exit(constants.EXIT_FAILURE) 
1122   
1123   
1124 -def PrepNoded(options, _): 
1125    """Preparation node daemon function, executed with the PID file held. 
1126   
1127    """ 
1128    if options.mlock: 
1129      request_executor_class = MlockallRequestExecutor 
1130      try: 
1131        utils.Mlockall() 
1132      except errors.NoCtypesError: 
1133        logging.warning("Cannot set memory lock, ctypes module not found") 
1134        request_executor_class = http.server.HttpServerRequestExecutor 
1135    else: 
1136      request_executor_class = http.server.HttpServerRequestExecutor 
1137   
1138    # Read SSL certificate 
1139    if options.ssl: 
1140      ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key, 
1141                                      ssl_cert_path=options.ssl_cert) 
1142    else: 
1143      ssl_params = None 
1144   
1145    err = _PrepareQueueLock() 
1146    if err is not None: 
1147      # this might be some kind of file-system/permission error; while 
1148      # this breaks the job queue functionality, we shouldn't prevent 
1149      # startup of the whole node daemon because of this 
1150      logging.critical("Can't init/verify the queue, proceeding anyway: %s", err) 
1151   
1152    handler = NodeRequestHandler() 
1153   
1154    mainloop = daemon.Mainloop() 
1155    server = \ 
1156      http.server.HttpServer(mainloop, options.bind_address, options.port, 
1157                             handler, ssl_params=ssl_params, ssl_verify_peer=True, 
1158                             request_executor_class=request_executor_class) 
1159    server.Start() 
1160   
1161    return (mainloop, server) 
1162   
1163   
1164 -def ExecNoded(options, args, prep_data): # pylint: disable=W0613 
1165    """Main node daemon function, executed with the PID file held. 
1166   
1167    """ 
1168    (mainloop, server) = prep_data 
1169    try: 
1170      mainloop.Run() 
1171    finally: 
1172      server.Stop() 
1173   
1174   
1175 -def Main(): 
1176    """Main function for the node daemon. 
1177   
1178    """ 
1179    parser = OptionParser(description="Ganeti node daemon", 
1180                          usage="%prog [-f] [-d] [-p port] [-b ADDRESS]", 
1181                          version="%%prog (ganeti) %s" % 
1182                          constants.RELEASE_VERSION) 
1183    parser.add_option("--no-mlock", dest="mlock", 
1184                      help="Do not mlock the node memory in ram", 
1185                      default=True, action="store_false") 
1186   
1187    daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded, 
1188                       default_ssl_cert=pathutils.NODED_CERT_FILE, 
1189                       default_ssl_key=pathutils.NODED_CERT_FILE, 
1190                       console_logging=True) 
1191