code.grnet.gr Git - ganeti-local/blob - lib/server/noded.py

   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.storage import container
  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) < 5:
 365       raise ValueError("Received only %s parameters in blockdev_grow,"
 366                        " old master?" % len(params))
 367     cfbd = objects.Disk.FromDict(params[0])
 368     amount = params[1]
 369     dryrun = params[2]
 370     backingstore = params[3]
 371     excl_stor = params[4]
 372     return backend.BlockdevGrow(cfbd, amount, dryrun, backingstore, excl_stor)
 373
 374   @staticmethod
 375   def perspective_blockdev_close(params):
 376     """Closes the given block devices.
 377
 378     """
 379     disks = [objects.Disk.FromDict(cf) for cf in params[1]]
 380     return backend.BlockdevClose(params[0], disks)
 381
 382   @staticmethod
 383   def perspective_blockdev_getdimensions(params):
 384     """Compute the sizes of the given block devices.
 385
 386     """
 387     disks = [objects.Disk.FromDict(cf) for cf in params[0]]
 388     return backend.BlockdevGetdimensions(disks)
 389
 390   @staticmethod
 391   def perspective_blockdev_export(params):
 392     """Compute the sizes of the given block devices.
 393
 394     """
 395     disk = objects.Disk.FromDict(params[0])
 396     dest_node, dest_path, cluster_name = params[1:]
 397     return backend.BlockdevExport(disk, dest_node, dest_path, cluster_name)
 398
 399   @staticmethod
 400   def perspective_blockdev_setinfo(params):
 401     """Sets metadata information on the given block device.
 402
 403     """
 404     (disk, info) = params
 405     disk = objects.Disk.FromDict(disk)
 406     return backend.BlockdevSetInfo(disk, info)
 407
 408   # blockdev/drbd specific methods ----------
 409
 410   @staticmethod
 411   def perspective_drbd_disconnect_net(params):
 412     """Disconnects the network connection of drbd disks.
 413
 414     Note that this is only valid for drbd disks, so the members of the
 415     disk list must all be drbd devices.
 416
 417     """
 418     nodes_ip, disks, target_node_uuid = params
 419     disks = [objects.Disk.FromDict(cf) for cf in disks]
 420     return backend.DrbdDisconnectNet(target_node_uuid, nodes_ip, disks)
 421
 422   @staticmethod
 423   def perspective_drbd_attach_net(params):
 424     """Attaches the network connection of drbd disks.
 425
 426     Note that this is only valid for drbd disks, so the members of the
 427     disk list must all be drbd devices.
 428
 429     """
 430     nodes_ip, disks, instance_name, multimaster, target_node_uuid = params
 431     disks = [objects.Disk.FromDict(cf) for cf in disks]
 432     return backend.DrbdAttachNet(target_node_uuid, nodes_ip, disks,
 433                                  instance_name, multimaster)
 434
 435   @staticmethod
 436   def perspective_drbd_wait_sync(params):
 437     """Wait until DRBD disks are synched.
 438
 439     Note that this is only valid for drbd disks, so the members of the
 440     disk list must all be drbd devices.
 441
 442     """
 443     nodes_ip, disks, target_node_uuid = params
 444     disks = [objects.Disk.FromDict(cf) for cf in disks]
 445     return backend.DrbdWaitSync(target_node_uuid, nodes_ip, disks)
 446
 447   @staticmethod
 448   def perspective_drbd_needs_activation(params):
 449     """Checks if the drbd devices need activation
 450
 451     Note that this is only valid for drbd disks, so the members of the
 452     disk list must all be drbd devices.
 453
 454     """
 455     nodes_ip, disks, target_node_uuid = params
 456     disks = [objects.Disk.FromDict(cf) for cf in disks]
 457     return backend.DrbdNeedsActivation(target_node_uuid, nodes_ip, disks)
 458
 459   @staticmethod
 460   def perspective_drbd_helper(params):
 461     """Query drbd helper.
 462
 463     """
 464     return backend.GetDrbdUsermodeHelper()
 465
 466   # export/import  --------------------------
 467
 468   @staticmethod
 469   def perspective_finalize_export(params):
 470     """Expose the finalize export functionality.
 471
 472     """
 473     instance = objects.Instance.FromDict(params[0])
 474
 475     snap_disks = []
 476     for disk in params[1]:
 477       if isinstance(disk, bool):
 478         snap_disks.append(disk)
 479       else:
 480         snap_disks.append(objects.Disk.FromDict(disk))
 481
 482     return backend.FinalizeExport(instance, snap_disks)
 483
 484   @staticmethod
 485   def perspective_export_info(params):
 486     """Query information about an existing export on this node.
 487
 488     The given path may not contain an export, in which case we return
 489     None.
 490
 491     """
 492     path = params[0]
 493     return backend.ExportInfo(path)
 494
 495   @staticmethod
 496   def perspective_export_list(params):
 497     """List the available exports on this node.
 498
 499     Note that as opposed to export_info, which may query data about an
 500     export in any path, this only queries the standard Ganeti path
 501     (pathutils.EXPORT_DIR).
 502
 503     """
 504     return backend.ListExports()
 505
 506   @staticmethod
 507   def perspective_export_remove(params):
 508     """Remove an export.
 509
 510     """
 511     export = params[0]
 512     return backend.RemoveExport(export)
 513
 514   # block device ---------------------
 515   @staticmethod
 516   def perspective_bdev_sizes(params):
 517     """Query the list of block devices
 518
 519     """
 520     devices = params[0]
 521     return backend.GetBlockDevSizes(devices)
 522
 523   # volume  --------------------------
 524
 525   @staticmethod
 526   def perspective_lv_list(params):
 527     """Query the list of logical volumes in a given volume group.
 528
 529     """
 530     vgname = params[0]
 531     return backend.GetVolumeList(vgname)
 532
 533   @staticmethod
 534   def perspective_vg_list(params):
 535     """Query the list of volume groups.
 536
 537     """
 538     return backend.ListVolumeGroups()
 539
 540   # Storage --------------------------
 541
 542   @staticmethod
 543   def perspective_storage_list(params):
 544     """Get list of storage units.
 545
 546     """
 547     (su_name, su_args, name, fields) = params
 548     return container.GetStorage(su_name, *su_args).List(name, fields)
 549
 550   @staticmethod
 551   def perspective_storage_modify(params):
 552     """Modify a storage unit.
 553
 554     """
 555     (su_name, su_args, name, changes) = params
 556     return container.GetStorage(su_name, *su_args).Modify(name, changes)
 557
 558   @staticmethod
 559   def perspective_storage_execute(params):
 560     """Execute an operation on a storage unit.
 561
 562     """
 563     (su_name, su_args, name, op) = params
 564     return container.GetStorage(su_name, *su_args).Execute(name, op)
 565
 566   # bridge  --------------------------
 567
 568   @staticmethod
 569   def perspective_bridges_exist(params):
 570     """Check if all bridges given exist on this node.
 571
 572     """
 573     bridges_list = params[0]
 574     return backend.BridgesExist(bridges_list)
 575
 576   # instance  --------------------------
 577
 578   @staticmethod
 579   def perspective_instance_os_add(params):
 580     """Install an OS on a given instance.
 581
 582     """
 583     inst_s = params[0]
 584     inst = objects.Instance.FromDict(inst_s)
 585     reinstall = params[1]
 586     debug = params[2]
 587     return backend.InstanceOsAdd(inst, reinstall, debug)
 588
 589   @staticmethod
 590   def perspective_instance_run_rename(params):
 591     """Runs the OS rename script for an instance.
 592
 593     """
 594     inst_s, old_name, debug = params
 595     inst = objects.Instance.FromDict(inst_s)
 596     return backend.RunRenameInstance(inst, old_name, debug)
 597
 598   @staticmethod
 599   def perspective_instance_shutdown(params):
 600     """Shutdown an instance.
 601
 602     """
 603     instance = objects.Instance.FromDict(params[0])
 604     timeout = params[1]
 605     trail = params[2]
 606     _extendReasonTrail(trail, "shutdown")
 607     return backend.InstanceShutdown(instance, timeout, trail)
 608
 609   @staticmethod
 610   def perspective_instance_start(params):
 611     """Start an instance.
 612
 613     """
 614     (instance_name, startup_paused, trail) = params
 615     instance = objects.Instance.FromDict(instance_name)
 616     _extendReasonTrail(trail, "start")
 617     return backend.StartInstance(instance, startup_paused, trail)
 618
 619   @staticmethod
 620   def perspective_migration_info(params):
 621     """Gather information about an instance to be migrated.
 622
 623     """
 624     instance = objects.Instance.FromDict(params[0])
 625     return backend.MigrationInfo(instance)
 626
 627   @staticmethod
 628   def perspective_accept_instance(params):
 629     """Prepare the node to accept an instance.
 630
 631     """
 632     instance, info, target = params
 633     instance = objects.Instance.FromDict(instance)
 634     return backend.AcceptInstance(instance, info, target)
 635
 636   @staticmethod
 637   def perspective_instance_finalize_migration_dst(params):
 638     """Finalize the instance migration on the destination node.
 639
 640     """
 641     instance, info, success = params
 642     instance = objects.Instance.FromDict(instance)
 643     return backend.FinalizeMigrationDst(instance, info, success)
 644
 645   @staticmethod
 646   def perspective_instance_migrate(params):
 647     """Migrates an instance.
 648
 649     """
 650     cluster_name, instance, target, live = params
 651     instance = objects.Instance.FromDict(instance)
 652     return backend.MigrateInstance(cluster_name, instance, target, live)
 653
 654   @staticmethod
 655   def perspective_instance_finalize_migration_src(params):
 656     """Finalize the instance migration on the source node.
 657
 658     """
 659     instance, success, live = params
 660     instance = objects.Instance.FromDict(instance)
 661     return backend.FinalizeMigrationSource(instance, success, live)
 662
 663   @staticmethod
 664   def perspective_instance_get_migration_status(params):
 665     """Reports migration status.
 666
 667     """
 668     instance = objects.Instance.FromDict(params[0])
 669     return backend.GetMigrationStatus(instance).ToDict()
 670
 671   @staticmethod
 672   def perspective_instance_reboot(params):
 673     """Reboot an instance.
 674
 675     """
 676     instance = objects.Instance.FromDict(params[0])
 677     reboot_type = params[1]
 678     shutdown_timeout = params[2]
 679     trail = params[3]
 680     _extendReasonTrail(trail, "reboot")
 681     return backend.InstanceReboot(instance, reboot_type, shutdown_timeout,
 682                                   trail)
 683
 684   @staticmethod
 685   def perspective_instance_balloon_memory(params):
 686     """Modify instance runtime memory.
 687
 688     """
 689     instance_dict, memory = params
 690     instance = objects.Instance.FromDict(instance_dict)
 691     return backend.InstanceBalloonMemory(instance, memory)
 692
 693   @staticmethod
 694   def perspective_instance_info(params):
 695     """Query instance information.
 696
 697     """
 698     (instance_name, hypervisor_name, hvparams) = params
 699     return backend.GetInstanceInfo(instance_name, hypervisor_name, hvparams)
 700
 701   @staticmethod
 702   def perspective_instance_migratable(params):
 703     """Query whether the specified instance can be migrated.
 704
 705     """
 706     instance = objects.Instance.FromDict(params[0])
 707     return backend.GetInstanceMigratable(instance)
 708
 709   @staticmethod
 710   def perspective_all_instances_info(params):
 711     """Query information about all instances.
 712
 713     """
 714     (hypervisor_list, all_hvparams) = params
 715     return backend.GetAllInstancesInfo(hypervisor_list, all_hvparams)
 716
 717   @staticmethod
 718   def perspective_instance_list(params):
 719     """Query the list of running instances.
 720
 721     """
 722     (hypervisor_list, hvparams) = params
 723     return backend.GetInstanceList(hypervisor_list, hvparams)
 724
 725   # node --------------------------
 726
 727   @staticmethod
 728   def perspective_node_has_ip_address(params):
 729     """Checks if a node has the given ip address.
 730
 731     """
 732     return netutils.IPAddress.Own(params[0])
 733
 734   @staticmethod
 735   def perspective_node_info(params):
 736     """Query node information.
 737
 738     """
 739     (storage_units, hv_specs) = params
 740     return backend.GetNodeInfo(storage_units, hv_specs)
 741
 742   @staticmethod
 743   def perspective_etc_hosts_modify(params):
 744     """Modify a node entry in /etc/hosts.
 745
 746     """
 747     backend.EtcHostsModify(params[0], params[1], params[2])
 748
 749     return True
 750
 751   @staticmethod
 752   def perspective_node_verify(params):
 753     """Run a verify sequence on this node.
 754
 755     """
 756     (what, cluster_name, hvparams) = params
 757     return backend.VerifyNode(what, cluster_name, hvparams)
 758
 759   @classmethod
 760   def perspective_node_verify_light(cls, params):
 761     """Run a light verify sequence on this node.
 762
 763     """
 764     # So far it's the same as the normal node_verify
 765     return cls.perspective_node_verify(params)
 766
 767   @staticmethod
 768   def perspective_node_start_master_daemons(params):
 769     """Start the master daemons on this node.
 770
 771     """
 772     return backend.StartMasterDaemons(params[0])
 773
 774   @staticmethod
 775   def perspective_node_activate_master_ip(params):
 776     """Activate the master IP on this node.
 777
 778     """
 779     master_params = objects.MasterNetworkParameters.FromDict(params[0])
 780     return backend.ActivateMasterIp(master_params, params[1])
 781
 782   @staticmethod
 783   def perspective_node_deactivate_master_ip(params):
 784     """Deactivate the master IP on this node.
 785
 786     """
 787     master_params = objects.MasterNetworkParameters.FromDict(params[0])
 788     return backend.DeactivateMasterIp(master_params, params[1])
 789
 790   @staticmethod
 791   def perspective_node_stop_master(params):
 792     """Stops master daemons on this node.
 793
 794     """
 795     return backend.StopMasterDaemons()
 796
 797   @staticmethod
 798   def perspective_node_change_master_netmask(params):
 799     """Change the master IP netmask.
 800
 801     """
 802     return backend.ChangeMasterNetmask(params[0], params[1], params[2],
 803                                        params[3])
 804
 805   @staticmethod
 806   def perspective_node_leave_cluster(params):
 807     """Cleanup after leaving a cluster.
 808
 809     """
 810     return backend.LeaveCluster(params[0])
 811
 812   @staticmethod
 813   def perspective_node_volumes(params):
 814     """Query the list of all logical volume groups.
 815
 816     """
 817     return backend.NodeVolumes()
 818
 819   @staticmethod
 820   def perspective_node_demote_from_mc(params):
 821     """Demote a node from the master candidate role.
 822
 823     """
 824     return backend.DemoteFromMC()
 825
 826   @staticmethod
 827   def perspective_node_powercycle(params):
 828     """Tries to powercycle the nod.
 829
 830     """
 831     (hypervisor_type, hvparams) = params
 832     return backend.PowercycleNode(hypervisor_type, hvparams)
 833
 834   # cluster --------------------------
 835
 836   @staticmethod
 837   def perspective_version(params):
 838     """Query version information.
 839
 840     """
 841     return constants.PROTOCOL_VERSION
 842
 843   @staticmethod
 844   def perspective_upload_file(params):
 845     """Upload a file.
 846
 847     Note that the backend implementation imposes strict rules on which
 848     files are accepted.
 849
 850     """
 851     return backend.UploadFile(*(params[0]))
 852
 853   @staticmethod
 854   def perspective_master_info(params):
 855     """Query master information.
 856
 857     """
 858     return backend.GetMasterInfo()
 859
 860   @staticmethod
 861   def perspective_run_oob(params):
 862     """Runs oob on node.
 863
 864     """
 865     output = backend.RunOob(params[0], params[1], params[2], params[3])
 866     if output:
 867       result = serializer.LoadJson(output)
 868     else:
 869       result = None
 870     return result
 871
 872   @staticmethod
 873   def perspective_restricted_command(params):
 874     """Runs a restricted command.
 875
 876     """
 877     (cmd, ) = params
 878
 879     return backend.RunRestrictedCmd(cmd)
 880
 881   @staticmethod
 882   def perspective_write_ssconf_files(params):
 883     """Write ssconf files.
 884
 885     """
 886     (values,) = params
 887     return ssconf.WriteSsconfFiles(values)
 888
 889   @staticmethod
 890   def perspective_get_watcher_pause(params):
 891     """Get watcher pause end.
 892
 893     """
 894     return utils.ReadWatcherPauseFile(pathutils.WATCHER_PAUSEFILE)
 895
 896   @staticmethod
 897   def perspective_set_watcher_pause(params):
 898     """Set watcher pause.
 899
 900     """
 901     (until, ) = params
 902     return backend.SetWatcherPause(until)
 903
 904   # os -----------------------
 905
 906   @staticmethod
 907   def perspective_os_diagnose(params):
 908     """Query detailed information about existing OSes.
 909
 910     """
 911     return backend.DiagnoseOS()
 912
 913   @staticmethod
 914   def perspective_os_get(params):
 915     """Query information about a given OS.
 916
 917     """
 918     name = params[0]
 919     os_obj = backend.OSFromDisk(name)
 920     return os_obj.ToDict()
 921
 922   @staticmethod
 923   def perspective_os_validate(params):
 924     """Run a given OS' validation routine.
 925
 926     """
 927     required, name, checks, params = params
 928     return backend.ValidateOS(required, name, checks, params)
 929
 930   # extstorage -----------------------
 931
 932   @staticmethod
 933   def perspective_extstorage_diagnose(params):
 934     """Query detailed information about existing extstorage providers.
 935
 936     """
 937     return backend.DiagnoseExtStorage()
 938
 939   # hooks -----------------------
 940
 941   @staticmethod
 942   def perspective_hooks_runner(params):
 943     """Run hook scripts.
 944
 945     """
 946     hpath, phase, env = params
 947     hr = backend.HooksRunner()
 948     return hr.RunHooks(hpath, phase, env)
 949
 950   # iallocator -----------------
 951
 952   @staticmethod
 953   def perspective_iallocator_runner(params):
 954     """Run an iallocator script.
 955
 956     """
 957     name, idata = params
 958     iar = backend.IAllocatorRunner()
 959     return iar.Run(name, idata)
 960
 961   # test -----------------------
 962
 963   @staticmethod
 964   def perspective_test_delay(params):
 965     """Run test delay.
 966
 967     """
 968     duration = params[0]
 969     status, rval = utils.TestDelay(duration)
 970     if not status:
 971       raise backend.RPCFail(rval)
 972     return rval
 973
 974   # file storage ---------------
 975
 976   @staticmethod
 977   def perspective_file_storage_dir_create(params):
 978     """Create the file storage directory.
 979
 980     """
 981     file_storage_dir = params[0]
 982     return backend.CreateFileStorageDir(file_storage_dir)
 983
 984   @staticmethod
 985   def perspective_file_storage_dir_remove(params):
 986     """Remove the file storage directory.
 987
 988     """
 989     file_storage_dir = params[0]
 990     return backend.RemoveFileStorageDir(file_storage_dir)
 991
 992   @staticmethod
 993   def perspective_file_storage_dir_rename(params):
 994     """Rename the file storage directory.
 995
 996     """
 997     old_file_storage_dir = params[0]
 998     new_file_storage_dir = params[1]
 999     return backend.RenameFileStorageDir(old_file_storage_dir,
1000                                         new_file_storage_dir)
1001
1002   # jobs ------------------------
1003
1004   @staticmethod
1005   @_RequireJobQueueLock
1006   def perspective_jobqueue_update(params):
1007     """Update job queue.
1008
1009     """
1010     (file_name, content) = params
1011     return backend.JobQueueUpdate(file_name, content)
1012
1013   @staticmethod
1014   @_RequireJobQueueLock
1015   def perspective_jobqueue_purge(params):
1016     """Purge job queue.
1017
1018     """
1019     return backend.JobQueuePurge()
1020
1021   @staticmethod
1022   @_RequireJobQueueLock
1023   def perspective_jobqueue_rename(params):
1024     """Rename a job queue file.
1025
1026     """
1027     # TODO: What if a file fails to rename?
1028     return [backend.JobQueueRename(old, new) for old, new in params[0]]
1029
1030   @staticmethod
1031   @_RequireJobQueueLock
1032   def perspective_jobqueue_set_drain_flag(params):
1033     """Set job queue's drain flag.
1034
1035     """
1036     (flag, ) = params
1037
1038     return jstore.SetDrainFlag(flag)
1039
1040   # hypervisor ---------------
1041
1042   @staticmethod
1043   def perspective_hypervisor_validate_params(params):
1044     """Validate the hypervisor parameters.
1045
1046     """
1047     (hvname, hvparams) = params
1048     return backend.ValidateHVParams(hvname, hvparams)
1049
1050   # Crypto
1051
1052   @staticmethod
1053   def perspective_x509_cert_create(params):
1054     """Creates a new X509 certificate for SSL/TLS.
1055
1056     """
1057     (validity, ) = params
1058     return backend.CreateX509Certificate(validity)
1059
1060   @staticmethod
1061   def perspective_x509_cert_remove(params):
1062     """Removes a X509 certificate.
1063
1064     """
1065     (name, ) = params
1066     return backend.RemoveX509Certificate(name)
1067
1068   # Import and export
1069
1070   @staticmethod
1071   def perspective_import_start(params):
1072     """Starts an import daemon.
1073
1074     """
1075     (opts_s, instance, component, (dest, dest_args)) = params
1076
1077     opts = objects.ImportExportOptions.FromDict(opts_s)
1078
1079     return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
1080                                            None, None,
1081                                            objects.Instance.FromDict(instance),
1082                                            component, dest,
1083                                            _DecodeImportExportIO(dest,
1084                                                                  dest_args))
1085
1086   @staticmethod
1087   def perspective_export_start(params):
1088     """Starts an export daemon.
1089
1090     """
1091     (opts_s, host, port, instance, component, (source, source_args)) = params
1092
1093     opts = objects.ImportExportOptions.FromDict(opts_s)
1094
1095     return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
1096                                            host, port,
1097                                            objects.Instance.FromDict(instance),
1098                                            component, source,
1099                                            _DecodeImportExportIO(source,
1100                                                                  source_args))
1101
1102   @staticmethod
1103   def perspective_impexp_status(params):
1104     """Retrieves the status of an import or export daemon.
1105
1106     """
1107     return backend.GetImportExportStatus(params[0])
1108
1109   @staticmethod
1110   def perspective_impexp_abort(params):
1111     """Aborts an import or export.
1112
1113     """
1114     return backend.AbortImportExport(params[0])
1115
1116   @staticmethod
1117   def perspective_impexp_cleanup(params):
1118     """Cleans up after an import or export.
1119
1120     """
1121     return backend.CleanupImportExport(params[0])
1122
1123
1124 def CheckNoded(_, args):
1125   """Initial checks whether to run or exit with a failure.
1126
1127   """
1128   if args: # noded doesn't take any arguments
1129     print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
1130                           sys.argv[0])
1131     sys.exit(constants.EXIT_FAILURE)
1132   try:
1133     codecs.lookup("string-escape")
1134   except LookupError:
1135     print >> sys.stderr, ("Can't load the string-escape code which is part"
1136                           " of the Python installation. Is your installation"
1137                           " complete/correct? Aborting.")
1138     sys.exit(constants.EXIT_FAILURE)
1139
1140
1141 def PrepNoded(options, _):
1142   """Preparation node daemon function, executed with the PID file held.
1143
1144   """
1145   if options.mlock:
1146     request_executor_class = MlockallRequestExecutor
1147     try:
1148       utils.Mlockall()
1149     except errors.NoCtypesError:
1150       logging.warning("Cannot set memory lock, ctypes module not found")
1151       request_executor_class = http.server.HttpServerRequestExecutor
1152   else:
1153     request_executor_class = http.server.HttpServerRequestExecutor
1154
1155   # Read SSL certificate
1156   if options.ssl:
1157     ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
1158                                     ssl_cert_path=options.ssl_cert)
1159   else:
1160     ssl_params = None
1161
1162   err = _PrepareQueueLock()
1163   if err is not None:
1164     # this might be some kind of file-system/permission error; while
1165     # this breaks the job queue functionality, we shouldn't prevent
1166     # startup of the whole node daemon because of this
1167     logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
1168
1169   handler = NodeRequestHandler()
1170
1171   mainloop = daemon.Mainloop()
1172   server = \
1173     http.server.HttpServer(mainloop, options.bind_address, options.port,
1174                            handler, ssl_params=ssl_params, ssl_verify_peer=True,
1175                            request_executor_class=request_executor_class)
1176   server.Start()
1177
1178   return (mainloop, server)
1179
1180
1181 def ExecNoded(options, args, prep_data): # pylint: disable=W0613
1182   """Main node daemon function, executed with the PID file held.
1183
1184   """
1185   (mainloop, server) = prep_data
1186   try:
1187     mainloop.Run()
1188   finally:
1189     server.Stop()
1190
1191
1192 def Main():
1193   """Main function for the node daemon.
1194
1195   """
1196   parser = OptionParser(description="Ganeti node daemon",
1197                         usage="%prog [-f] [-d] [-p port] [-b ADDRESS]\
1198                               \ [-i INTERFACE]",
1199                         version="%%prog (ganeti) %s" %
1200                         constants.RELEASE_VERSION)
1201   parser.add_option("--no-mlock", dest="mlock",
1202                     help="Do not mlock the node memory in ram",
1203                     default=True, action="store_false")
1204
1205   daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded,
1206                      default_ssl_cert=pathutils.NODED_CERT_FILE,
1207                      default_ssl_key=pathutils.NODED_CERT_FILE,
1208                      console_logging=True)