code.grnet.gr Git - ganeti-local/blob - daemons/ganeti-noded

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2006, 2007 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 # functions in this module need to have a given name structure, so:
  25 # pylint: disable-msg=C0103
  26
  27 import os
  28 import sys
  29 import traceback
  30 import SocketServer
  31 import errno
  32 import logging
  33 import signal
  34
  35 from optparse import OptionParser
  36
  37 from ganeti import backend
  38 from ganeti import constants
  39 from ganeti import objects
  40 from ganeti import errors
  41 from ganeti import jstore
  42 from ganeti import daemon
  43 from ganeti import http
  44 from ganeti import utils
  45
  46 import ganeti.http.server
  47
  48
  49 queue_lock = None
  50
  51
  52 def _RequireJobQueueLock(fn):
  53   """Decorator for job queue manipulating functions.
  54
  55   """
  56   QUEUE_LOCK_TIMEOUT = 10
  57
  58   def wrapper(*args, **kwargs):
  59     # Locking in exclusive, blocking mode because there could be several
  60     # children running at the same time. Waiting up to 10 seconds.
  61     queue_lock.Exclusive(blocking=True, timeout=QUEUE_LOCK_TIMEOUT)
  62     try:
  63       return fn(*args, **kwargs)
  64     finally:
  65       queue_lock.Unlock()
  66
  67   return wrapper
  68
  69
  70 class NodeHttpServer(http.server.HttpServer):
  71   """The server implementation.
  72
  73   This class holds all methods exposed over the RPC interface.
  74
  75   """
  76   def __init__(self, *args, **kwargs):
  77     http.server.HttpServer.__init__(self, *args, **kwargs)
  78     self.noded_pid = os.getpid()
  79
  80   def HandleRequest(self, req):
  81     """Handle a request.
  82
  83     """
  84     if req.request_method.upper() != http.HTTP_PUT:
  85       raise http.HttpBadRequest()
  86
  87     path = req.request_path
  88     if path.startswith("/"):
  89       path = path[1:]
  90
  91     method = getattr(self, "perspective_%s" % path, None)
  92     if method is None:
  93       raise http.HttpNotFound()
  94
  95     try:
  96       try:
  97         return method(req.request_body)
  98       except:
  99         logging.exception("Error in RPC call")
 100         raise
 101     except errors.QuitGanetiException, err:
 102       # Tell parent to quit
 103       os.kill(self.noded_pid, signal.SIGTERM)
 104
 105   # the new block devices  --------------------------
 106
 107   @staticmethod
 108   def perspective_blockdev_create(params):
 109     """Create a block device.
 110
 111     """
 112     bdev_s, size, owner, on_primary, info = params
 113     bdev = objects.Disk.FromDict(bdev_s)
 114     if bdev is None:
 115       raise ValueError("can't unserialize data!")
 116     return backend.CreateBlockDevice(bdev, size, owner, on_primary, info)
 117
 118   @staticmethod
 119   def perspective_blockdev_remove(params):
 120     """Remove a block device.
 121
 122     """
 123     bdev_s = params[0]
 124     bdev = objects.Disk.FromDict(bdev_s)
 125     return backend.RemoveBlockDevice(bdev)
 126
 127   @staticmethod
 128   def perspective_blockdev_rename(params):
 129     """Remove a block device.
 130
 131     """
 132     devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params]
 133     return backend.RenameBlockDevices(devlist)
 134
 135   @staticmethod
 136   def perspective_blockdev_assemble(params):
 137     """Assemble a block device.
 138
 139     """
 140     bdev_s, owner, on_primary = params
 141     bdev = objects.Disk.FromDict(bdev_s)
 142     if bdev is None:
 143       raise ValueError("can't unserialize data!")
 144     return backend.AssembleBlockDevice(bdev, owner, on_primary)
 145
 146   @staticmethod
 147   def perspective_blockdev_shutdown(params):
 148     """Shutdown a block device.
 149
 150     """
 151     bdev_s = params[0]
 152     bdev = objects.Disk.FromDict(bdev_s)
 153     if bdev is None:
 154       raise ValueError("can't unserialize data!")
 155     return backend.ShutdownBlockDevice(bdev)
 156
 157   @staticmethod
 158   def perspective_blockdev_addchildren(params):
 159     """Add a child to a mirror device.
 160
 161     Note: this is only valid for mirror devices. It's the caller's duty
 162     to send a correct disk, otherwise we raise an error.
 163
 164     """
 165     bdev_s, ndev_s = params
 166     bdev = objects.Disk.FromDict(bdev_s)
 167     ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
 168     if bdev is None or ndevs.count(None) > 0:
 169       raise ValueError("can't unserialize data!")
 170     return backend.MirrorAddChildren(bdev, ndevs)
 171
 172   @staticmethod
 173   def perspective_blockdev_removechildren(params):
 174     """Remove a child from a mirror device.
 175
 176     This is only valid for mirror devices, of course. It's the callers
 177     duty to send a correct disk, otherwise we raise an error.
 178
 179     """
 180     bdev_s, ndev_s = params
 181     bdev = objects.Disk.FromDict(bdev_s)
 182     ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
 183     if bdev is None or ndevs.count(None) > 0:
 184       raise ValueError("can't unserialize data!")
 185     return backend.MirrorRemoveChildren(bdev, ndevs)
 186
 187   @staticmethod
 188   def perspective_blockdev_getmirrorstatus(params):
 189     """Return the mirror status for a list of disks.
 190
 191     """
 192     disks = [objects.Disk.FromDict(dsk_s)
 193             for dsk_s in params]
 194     return backend.GetMirrorStatus(disks)
 195
 196   @staticmethod
 197   def perspective_blockdev_find(params):
 198     """Expose the FindBlockDevice functionality for a disk.
 199
 200     This will try to find but not activate a disk.
 201
 202     """
 203     disk = objects.Disk.FromDict(params[0])
 204     return backend.FindBlockDevice(disk)
 205
 206   @staticmethod
 207   def perspective_blockdev_snapshot(params):
 208     """Create a snapshot device.
 209
 210     Note that this is only valid for LVM disks, if we get passed
 211     something else we raise an exception. The snapshot device can be
 212     remove by calling the generic block device remove call.
 213
 214     """
 215     cfbd = objects.Disk.FromDict(params[0])
 216     return backend.SnapshotBlockDevice(cfbd)
 217
 218   @staticmethod
 219   def perspective_blockdev_grow(params):
 220     """Grow a stack of devices.
 221
 222     """
 223     cfbd = objects.Disk.FromDict(params[0])
 224     amount = params[1]
 225     return backend.GrowBlockDevice(cfbd, amount)
 226
 227   @staticmethod
 228   def perspective_blockdev_close(params):
 229     """Closes the given block devices.
 230
 231     """
 232     disks = [objects.Disk.FromDict(cf) for cf in params]
 233     return backend.CloseBlockDevices(disks)
 234
 235   # export/import  --------------------------
 236
 237   @staticmethod
 238   def perspective_snapshot_export(params):
 239     """Export a given snapshot.
 240
 241     """
 242     disk = objects.Disk.FromDict(params[0])
 243     dest_node = params[1]
 244     instance = objects.Instance.FromDict(params[2])
 245     cluster_name = params[3]
 246     dev_idx = params[4]
 247     return backend.ExportSnapshot(disk, dest_node, instance,
 248                                   cluster_name, dev_idx)
 249
 250   @staticmethod
 251   def perspective_finalize_export(params):
 252     """Expose the finalize export functionality.
 253
 254     """
 255     instance = objects.Instance.FromDict(params[0])
 256     snap_disks = [objects.Disk.FromDict(str_data)
 257                   for str_data in params[1]]
 258     return backend.FinalizeExport(instance, snap_disks)
 259
 260   @staticmethod
 261   def perspective_export_info(params):
 262     """Query information about an existing export on this node.
 263
 264     The given path may not contain an export, in which case we return
 265     None.
 266
 267     """
 268     path = params[0]
 269     einfo = backend.ExportInfo(path)
 270     if einfo is None:
 271       return einfo
 272     return einfo.Dumps()
 273
 274   @staticmethod
 275   def perspective_export_list(params):
 276     """List the available exports on this node.
 277
 278     Note that as opposed to export_info, which may query data about an
 279     export in any path, this only queries the standard Ganeti path
 280     (constants.EXPORT_DIR).
 281
 282     """
 283     return backend.ListExports()
 284
 285   @staticmethod
 286   def perspective_export_remove(params):
 287     """Remove an export.
 288
 289     """
 290     export = params[0]
 291     return backend.RemoveExport(export)
 292
 293   # volume  --------------------------
 294
 295   @staticmethod
 296   def perspective_volume_list(params):
 297     """Query the list of logical volumes in a given volume group.
 298
 299     """
 300     vgname = params[0]
 301     return backend.GetVolumeList(vgname)
 302
 303   @staticmethod
 304   def perspective_vg_list(params):
 305     """Query the list of volume groups.
 306
 307     """
 308     return backend.ListVolumeGroups()
 309
 310   # bridge  --------------------------
 311
 312   @staticmethod
 313   def perspective_bridges_exist(params):
 314     """Check if all bridges given exist on this node.
 315
 316     """
 317     bridges_list = params[0]
 318     return backend.BridgesExist(bridges_list)
 319
 320   # instance  --------------------------
 321
 322   @staticmethod
 323   def perspective_instance_os_add(params):
 324     """Install an OS on a given instance.
 325
 326     """
 327     inst_s = params[0]
 328     inst = objects.Instance.FromDict(inst_s)
 329     return backend.AddOSToInstance(inst)
 330
 331   @staticmethod
 332   def perspective_instance_run_rename(params):
 333     """Runs the OS rename script for an instance.
 334
 335     """
 336     inst_s, old_name = params
 337     inst = objects.Instance.FromDict(inst_s)
 338     return backend.RunRenameInstance(inst, old_name)
 339
 340   @staticmethod
 341   def perspective_instance_os_import(params):
 342     """Run the import function of an OS onto a given instance.
 343
 344     """
 345     inst_s, src_node, src_images, cluster_name = params
 346     inst = objects.Instance.FromDict(inst_s)
 347     return backend.ImportOSIntoInstance(inst, src_node, src_images,
 348                                         cluster_name)
 349
 350   @staticmethod
 351   def perspective_instance_shutdown(params):
 352     """Shutdown an instance.
 353
 354     """
 355     instance = objects.Instance.FromDict(params[0])
 356     return backend.ShutdownInstance(instance)
 357
 358   @staticmethod
 359   def perspective_instance_start(params):
 360     """Start an instance.
 361
 362     """
 363     instance = objects.Instance.FromDict(params[0])
 364     extra_args = params[1]
 365     return backend.StartInstance(instance, extra_args)
 366
 367   @staticmethod
 368   def perspective_instance_migrate(params):
 369     """Migrates an instance.
 370
 371     """
 372     instance, target, live = params
 373     instance = objects.Instance.FromDict(instance)
 374     return backend.MigrateInstance(instance, target, live)
 375
 376   @staticmethod
 377   def perspective_instance_reboot(params):
 378     """Reboot an instance.
 379
 380     """
 381     instance = objects.Instance.FromDict(params[0])
 382     reboot_type = params[1]
 383     extra_args = params[2]
 384     return backend.RebootInstance(instance, reboot_type, extra_args)
 385
 386   @staticmethod
 387   def perspective_instance_info(params):
 388     """Query instance information.
 389
 390     """
 391     return backend.GetInstanceInfo(params[0], params[1])
 392
 393   @staticmethod
 394   def perspective_all_instances_info(params):
 395     """Query information about all instances.
 396
 397     """
 398     return backend.GetAllInstancesInfo(params[0])
 399
 400   @staticmethod
 401   def perspective_instance_list(params):
 402     """Query the list of running instances.
 403
 404     """
 405     return backend.GetInstanceList(params[0])
 406
 407   # node --------------------------
 408
 409   @staticmethod
 410   def perspective_node_tcp_ping(params):
 411     """Do a TcpPing on the remote node.
 412
 413     """
 414     return utils.TcpPing(params[1], params[2], timeout=params[3],
 415                          live_port_needed=params[4], source=params[0])
 416
 417   @staticmethod
 418   def perspective_node_has_ip_address(params):
 419     """Checks if a node has the given ip address.
 420
 421     """
 422     return utils.OwnIpAddress(params[0])
 423
 424   @staticmethod
 425   def perspective_node_info(params):
 426     """Query node information.
 427
 428     """
 429     vgname, hypervisor_type = params
 430     return backend.GetNodeInfo(vgname, hypervisor_type)
 431
 432   @staticmethod
 433   def perspective_node_add(params):
 434     """Complete the registration of this node in the cluster.
 435
 436     """
 437     return backend.AddNode(params[0], params[1], params[2],
 438                            params[3], params[4], params[5])
 439
 440   @staticmethod
 441   def perspective_node_verify(params):
 442     """Run a verify sequence on this node.
 443
 444     """
 445     return backend.VerifyNode(params[0], params[1])
 446
 447   @staticmethod
 448   def perspective_node_start_master(params):
 449     """Promote this node to master status.
 450
 451     """
 452     return backend.StartMaster(params[0])
 453
 454   @staticmethod
 455   def perspective_node_stop_master(params):
 456     """Demote this node from master status.
 457
 458     """
 459     return backend.StopMaster(params[0])
 460
 461   @staticmethod
 462   def perspective_node_leave_cluster(params):
 463     """Cleanup after leaving a cluster.
 464
 465     """
 466     return backend.LeaveCluster()
 467
 468   @staticmethod
 469   def perspective_node_volumes(params):
 470     """Query the list of all logical volume groups.
 471
 472     """
 473     return backend.NodeVolumes()
 474
 475   # cluster --------------------------
 476
 477   @staticmethod
 478   def perspective_version(params):
 479     """Query version information.
 480
 481     """
 482     return constants.PROTOCOL_VERSION
 483
 484   @staticmethod
 485   def perspective_upload_file(params):
 486     """Upload a file.
 487
 488     Note that the backend implementation imposes strict rules on which
 489     files are accepted.
 490
 491     """
 492     return backend.UploadFile(*params)
 493
 494   @staticmethod
 495   def perspective_master_info(params):
 496     """Query master information.
 497
 498     """
 499     return backend.GetMasterInfo()
 500
 501   @staticmethod
 502   def perspective_write_ssconf_files(params):
 503     """Write ssconf files.
 504
 505     """
 506     (values,) = params
 507     return backend.WriteSsconfFiles(values)
 508
 509   # os -----------------------
 510
 511   @staticmethod
 512   def perspective_os_diagnose(params):
 513     """Query detailed information about existing OSes.
 514
 515     """
 516     return [os.ToDict() for os in backend.DiagnoseOS()]
 517
 518   @staticmethod
 519   def perspective_os_get(params):
 520     """Query information about a given OS.
 521
 522     """
 523     name = params[0]
 524     try:
 525       os_obj = backend.OSFromDisk(name)
 526     except errors.InvalidOS, err:
 527       os_obj = objects.OS.FromInvalidOS(err)
 528     return os_obj.ToDict()
 529
 530   # hooks -----------------------
 531
 532   @staticmethod
 533   def perspective_hooks_runner(params):
 534     """Run hook scripts.
 535
 536     """
 537     hpath, phase, env = params
 538     hr = backend.HooksRunner()
 539     return hr.RunHooks(hpath, phase, env)
 540
 541   # iallocator -----------------
 542
 543   @staticmethod
 544   def perspective_iallocator_runner(params):
 545     """Run an iallocator script.
 546
 547     """
 548     name, idata = params
 549     iar = backend.IAllocatorRunner()
 550     return iar.Run(name, idata)
 551
 552   # test -----------------------
 553
 554   @staticmethod
 555   def perspective_test_delay(params):
 556     """Run test delay.
 557
 558     """
 559     duration = params[0]
 560     return utils.TestDelay(duration)
 561
 562   # file storage ---------------
 563
 564   @staticmethod
 565   def perspective_file_storage_dir_create(params):
 566     """Create the file storage directory.
 567
 568     """
 569     file_storage_dir = params[0]
 570     return backend.CreateFileStorageDir(file_storage_dir)
 571
 572   @staticmethod
 573   def perspective_file_storage_dir_remove(params):
 574     """Remove the file storage directory.
 575
 576     """
 577     file_storage_dir = params[0]
 578     return backend.RemoveFileStorageDir(file_storage_dir)
 579
 580   @staticmethod
 581   def perspective_file_storage_dir_rename(params):
 582     """Rename the file storage directory.
 583
 584     """
 585     old_file_storage_dir = params[0]
 586     new_file_storage_dir = params[1]
 587     return backend.RenameFileStorageDir(old_file_storage_dir,
 588                                         new_file_storage_dir)
 589
 590   # jobs ------------------------
 591
 592   @staticmethod
 593   @_RequireJobQueueLock
 594   def perspective_jobqueue_update(params):
 595     """Update job queue.
 596
 597     """
 598     (file_name, content) = params
 599     return backend.JobQueueUpdate(file_name, content)
 600
 601   @staticmethod
 602   @_RequireJobQueueLock
 603   def perspective_jobqueue_purge(params):
 604     """Purge job queue.
 605
 606     """
 607     return backend.JobQueuePurge()
 608
 609   @staticmethod
 610   @_RequireJobQueueLock
 611   def perspective_jobqueue_rename(params):
 612     """Rename a job queue file.
 613
 614     """
 615     (old, new) = params
 616
 617     return backend.JobQueueRename(old, new)
 618
 619   @staticmethod
 620   def perspective_jobqueue_set_drain(params):
 621     """Set/unset the queue drain flag.
 622
 623     """
 624     drain_flag = params[0]
 625     return backend.JobQueueSetDrainFlag(drain_flag)
 626
 627
 628   # hypervisor ---------------
 629
 630   @staticmethod
 631   def perspective_hypervisor_validate_params(params):
 632     """Validate the hypervisor parameters.
 633
 634     """
 635     (hvname, hvparams) = params
 636     return backend.ValidateHVParams(hvname, hvparams)
 637
 638
 639 def ParseOptions():
 640   """Parse the command line options.
 641
 642   Returns:
 643     (options, args) as from OptionParser.parse_args()
 644
 645   """
 646   parser = OptionParser(description="Ganeti node daemon",
 647                         usage="%prog [-f] [-d]",
 648                         version="%%prog (ganeti) %s" %
 649                         constants.RELEASE_VERSION)
 650
 651   parser.add_option("-f", "--foreground", dest="fork",
 652                     help="Don't detach from the current terminal",
 653                     default=True, action="store_false")
 654   parser.add_option("-d", "--debug", dest="debug",
 655                     help="Enable some debug messages",
 656                     default=False, action="store_true")
 657   options, args = parser.parse_args()
 658   return options, args
 659
 660
 661 def EnsureRuntimeEnvironment():
 662   """Ensure our run-time environment is complete.
 663
 664   Currently this creates directories which could be missing, either
 665   due to directories being on a tmpfs mount, or due to incomplete
 666   packaging.
 667
 668   """
 669   dirs = [(val, constants.RUN_DIRS_MODE) for val in constants.SUB_RUN_DIRS]
 670   dirs.append((constants.LOG_OS_DIR, 0750))
 671   for dir_name, dir_mode in dirs:
 672     if not os.path.exists(dir_name):
 673       try:
 674         os.mkdir(dir_name, dir_mode)
 675       except EnvironmentError, err:
 676         if err.errno != errno.EEXIST:
 677           print ("Node setup wrong, cannot create directory '%s': %s" %
 678                  (dir_name, err))
 679           sys.exit(5)
 680     if not os.path.isdir(dir_name):
 681       print ("Node setup wrong, '%s' is not a directory" % dir_name)
 682       sys.exit(5)
 683
 684
 685 def main():
 686   """Main function for the node daemon.
 687
 688   """
 689   global queue_lock
 690
 691   options, args = ParseOptions()
 692   utils.debug = options.debug
 693   for fname in (constants.SSL_CERT_FILE,):
 694     if not os.path.isfile(fname):
 695       print "config %s not there, will not run." % fname
 696       sys.exit(5)
 697
 698   try:
 699     port = utils.GetNodeDaemonPort()
 700   except errors.ConfigurationError, err:
 701     print "Cluster configuration incomplete: '%s'" % str(err)
 702     sys.exit(5)
 703
 704   EnsureRuntimeEnvironment()
 705
 706   # become a daemon
 707   if options.fork:
 708     utils.Daemonize(logfile=constants.LOG_NODESERVER)
 709
 710   utils.WritePidFile(constants.NODED_PID)
 711   try:
 712     utils.SetupLogging(logfile=constants.LOG_NODESERVER, debug=options.debug,
 713                        stderr_logging=not options.fork)
 714     logging.info("ganeti node daemon startup")
 715
 716     # Read SSL certificate
 717     ssl_params = http.HttpSslParams(ssl_key_path=constants.SSL_CERT_FILE,
 718                                     ssl_cert_path=constants.SSL_CERT_FILE)
 719
 720     # Prepare job queue
 721     queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
 722
 723     mainloop = daemon.Mainloop()
 724     server = NodeHttpServer(mainloop, "", port,
 725                             ssl_params=ssl_params, ssl_verify_peer=True)
 726     server.Start()
 727     try:
 728       mainloop.Run()
 729     finally:
 730       server.Stop()
 731   finally:
 732     utils.RemovePidFile(constants.NODED_PID)
 733
 734
 735 if __name__ == '__main__':
 736   main()