Statistics
| Branch: | Tag: | Revision:

root / daemons / ganeti-noded @ dd875d32

History | View | Annotate | Download (18.9 kB)

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
  @staticmethod
476
  def perspective_node_demote_from_mc(params):
477
    """Demote a node from the master candidate role.
478

    
479
    """
480
    return backend.DemoteFromMC()
481

    
482

    
483
  # cluster --------------------------
484

    
485
  @staticmethod
486
  def perspective_version(params):
487
    """Query version information.
488

    
489
    """
490
    return constants.PROTOCOL_VERSION
491

    
492
  @staticmethod
493
  def perspective_upload_file(params):
494
    """Upload a file.
495

    
496
    Note that the backend implementation imposes strict rules on which
497
    files are accepted.
498

    
499
    """
500
    return backend.UploadFile(*params)
501

    
502
  @staticmethod
503
  def perspective_master_info(params):
504
    """Query master information.
505

    
506
    """
507
    return backend.GetMasterInfo()
508

    
509
  @staticmethod
510
  def perspective_write_ssconf_files(params):
511
    """Write ssconf files.
512

    
513
    """
514
    (values,) = params
515
    return backend.WriteSsconfFiles(values)
516

    
517
  # os -----------------------
518

    
519
  @staticmethod
520
  def perspective_os_diagnose(params):
521
    """Query detailed information about existing OSes.
522

    
523
    """
524
    return [os.ToDict() for os in backend.DiagnoseOS()]
525

    
526
  @staticmethod
527
  def perspective_os_get(params):
528
    """Query information about a given OS.
529

    
530
    """
531
    name = params[0]
532
    try:
533
      os_obj = backend.OSFromDisk(name)
534
    except errors.InvalidOS, err:
535
      os_obj = objects.OS.FromInvalidOS(err)
536
    return os_obj.ToDict()
537

    
538
  # hooks -----------------------
539

    
540
  @staticmethod
541
  def perspective_hooks_runner(params):
542
    """Run hook scripts.
543

    
544
    """
545
    hpath, phase, env = params
546
    hr = backend.HooksRunner()
547
    return hr.RunHooks(hpath, phase, env)
548

    
549
  # iallocator -----------------
550

    
551
  @staticmethod
552
  def perspective_iallocator_runner(params):
553
    """Run an iallocator script.
554

    
555
    """
556
    name, idata = params
557
    iar = backend.IAllocatorRunner()
558
    return iar.Run(name, idata)
559

    
560
  # test -----------------------
561

    
562
  @staticmethod
563
  def perspective_test_delay(params):
564
    """Run test delay.
565

    
566
    """
567
    duration = params[0]
568
    return utils.TestDelay(duration)
569

    
570
  # file storage ---------------
571

    
572
  @staticmethod
573
  def perspective_file_storage_dir_create(params):
574
    """Create the file storage directory.
575

    
576
    """
577
    file_storage_dir = params[0]
578
    return backend.CreateFileStorageDir(file_storage_dir)
579

    
580
  @staticmethod
581
  def perspective_file_storage_dir_remove(params):
582
    """Remove the file storage directory.
583

    
584
    """
585
    file_storage_dir = params[0]
586
    return backend.RemoveFileStorageDir(file_storage_dir)
587

    
588
  @staticmethod
589
  def perspective_file_storage_dir_rename(params):
590
    """Rename the file storage directory.
591

    
592
    """
593
    old_file_storage_dir = params[0]
594
    new_file_storage_dir = params[1]
595
    return backend.RenameFileStorageDir(old_file_storage_dir,
596
                                        new_file_storage_dir)
597

    
598
  # jobs ------------------------
599

    
600
  @staticmethod
601
  @_RequireJobQueueLock
602
  def perspective_jobqueue_update(params):
603
    """Update job queue.
604

    
605
    """
606
    (file_name, content) = params
607
    return backend.JobQueueUpdate(file_name, content)
608

    
609
  @staticmethod
610
  @_RequireJobQueueLock
611
  def perspective_jobqueue_purge(params):
612
    """Purge job queue.
613

    
614
    """
615
    return backend.JobQueuePurge()
616

    
617
  @staticmethod
618
  @_RequireJobQueueLock
619
  def perspective_jobqueue_rename(params):
620
    """Rename a job queue file.
621

    
622
    """
623
    # TODO: What if a file fails to rename?
624
    return [backend.JobQueueRename(old, new) for old, new in params]
625

    
626
  @staticmethod
627
  def perspective_jobqueue_set_drain(params):
628
    """Set/unset the queue drain flag.
629

    
630
    """
631
    drain_flag = params[0]
632
    return backend.JobQueueSetDrainFlag(drain_flag)
633

    
634

    
635
  # hypervisor ---------------
636

    
637
  @staticmethod
638
  def perspective_hypervisor_validate_params(params):
639
    """Validate the hypervisor parameters.
640

    
641
    """
642
    (hvname, hvparams) = params
643
    return backend.ValidateHVParams(hvname, hvparams)
644

    
645

    
646
def ParseOptions():
647
  """Parse the command line options.
648

    
649
  @return: (options, args) as from OptionParser.parse_args()
650

    
651
  """
652
  parser = OptionParser(description="Ganeti node daemon",
653
                        usage="%prog [-f] [-d]",
654
                        version="%%prog (ganeti) %s" %
655
                        constants.RELEASE_VERSION)
656

    
657
  parser.add_option("-f", "--foreground", dest="fork",
658
                    help="Don't detach from the current terminal",
659
                    default=True, action="store_false")
660
  parser.add_option("-d", "--debug", dest="debug",
661
                    help="Enable some debug messages",
662
                    default=False, action="store_true")
663
  options, args = parser.parse_args()
664
  return options, args
665

    
666

    
667
def EnsureRuntimeEnvironment():
668
  """Ensure our run-time environment is complete.
669

    
670
  Currently this creates directories which could be missing, either
671
  due to directories being on a tmpfs mount, or due to incomplete
672
  packaging.
673

    
674
  """
675
  dirs = [(val, constants.RUN_DIRS_MODE) for val in constants.SUB_RUN_DIRS]
676
  dirs.append((constants.LOG_OS_DIR, 0750))
677
  for dir_name, dir_mode in dirs:
678
    if not os.path.exists(dir_name):
679
      try:
680
        os.mkdir(dir_name, dir_mode)
681
      except EnvironmentError, err:
682
        if err.errno != errno.EEXIST:
683
          print ("Node setup wrong, cannot create directory '%s': %s" %
684
                 (dir_name, err))
685
          sys.exit(5)
686
    if not os.path.isdir(dir_name):
687
      print ("Node setup wrong, '%s' is not a directory" % dir_name)
688
      sys.exit(5)
689

    
690

    
691
def main():
692
  """Main function for the node daemon.
693

    
694
  """
695
  global queue_lock
696

    
697
  options, args = ParseOptions()
698
  utils.debug = options.debug
699
  for fname in (constants.SSL_CERT_FILE,):
700
    if not os.path.isfile(fname):
701
      print "config %s not there, will not run." % fname
702
      sys.exit(5)
703

    
704
  try:
705
    port = utils.GetNodeDaemonPort()
706
  except errors.ConfigurationError, err:
707
    print "Cluster configuration incomplete: '%s'" % str(err)
708
    sys.exit(5)
709

    
710
  EnsureRuntimeEnvironment()
711

    
712
  # become a daemon
713
  if options.fork:
714
    utils.Daemonize(logfile=constants.LOG_NODESERVER)
715

    
716
  utils.WritePidFile(constants.NODED_PID)
717
  try:
718
    utils.SetupLogging(logfile=constants.LOG_NODESERVER, debug=options.debug,
719
                       stderr_logging=not options.fork)
720
    logging.info("ganeti node daemon startup")
721

    
722
    # Read SSL certificate
723
    ssl_params = http.HttpSslParams(ssl_key_path=constants.SSL_CERT_FILE,
724
                                    ssl_cert_path=constants.SSL_CERT_FILE)
725

    
726
    # Prepare job queue
727
    queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
728

    
729
    mainloop = daemon.Mainloop()
730
    server = NodeHttpServer(mainloop, "", port,
731
                            ssl_params=ssl_params, ssl_verify_peer=True)
732
    server.Start()
733
    try:
734
      mainloop.Run()
735
    finally:
736
      server.Stop()
737
  finally:
738
    utils.RemovePidFile(constants.NODED_PID)
739

    
740

    
741
if __name__ == '__main__':
742
  main()