Statistics
| Branch: | Tag: | Revision:

root / lib / server / noded.py @ 31135ddd

History | View | Annotate | Download (31.7 kB)

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 _PrepareQueueLock():
61
  """Try to prepare the queue lock.
62

63
  @return: None for success, otherwise an exception object
64

65
  """
66
  global queue_lock # pylint: disable=W0603
67

    
68
  if queue_lock is not None:
69
    return None
70

    
71
  # Prepare job queue
72
  try:
73
    queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
74
    return None
75
  except EnvironmentError, err:
76
    return err
77

    
78

    
79
def _RequireJobQueueLock(fn):
80
  """Decorator for job queue manipulating functions.
81

82
  """
83
  QUEUE_LOCK_TIMEOUT = 10
84

    
85
  def wrapper(*args, **kwargs):
86
    # Locking in exclusive, blocking mode because there could be several
87
    # children running at the same time. Waiting up to 10 seconds.
88
    if _PrepareQueueLock() is not None:
89
      raise errors.JobQueueError("Job queue failed initialization,"
90
                                 " cannot update jobs")
91
    queue_lock.Exclusive(blocking=True, timeout=QUEUE_LOCK_TIMEOUT)
92
    try:
93
      return fn(*args, **kwargs)
94
    finally:
95
      queue_lock.Unlock()
96

    
97
  return wrapper
98

    
99

    
100
def _DecodeImportExportIO(ieio, ieioargs):
101
  """Decodes import/export I/O information.
102

103
  """
104
  if ieio == constants.IEIO_RAW_DISK:
105
    assert len(ieioargs) == 1
106
    return (objects.Disk.FromDict(ieioargs[0]), )
107

    
108
  if ieio == constants.IEIO_SCRIPT:
109
    assert len(ieioargs) == 2
110
    return (objects.Disk.FromDict(ieioargs[0]), ieioargs[1])
111

    
112
  return ieioargs
113

    
114

    
115
class MlockallRequestExecutor(http.server.HttpServerRequestExecutor):
116
  """Subclass ensuring request handlers are locked in RAM.
117

118
  """
119
  def __init__(self, *args, **kwargs):
120
    utils.Mlockall()
121

    
122
    http.server.HttpServerRequestExecutor.__init__(self, *args, **kwargs)
123

    
124

    
125
class NodeRequestHandler(http.server.HttpServerHandler):
126
  """The server implementation.
127

128
  This class holds all methods exposed over the RPC interface.
129

130
  """
131
  # too many public methods, and unused args - all methods get params
132
  # due to the API
133
  # pylint: disable=R0904,W0613
134
  def __init__(self):
135
    http.server.HttpServerHandler.__init__(self)
136
    self.noded_pid = os.getpid()
137

    
138
  def HandleRequest(self, req):
139
    """Handle a request.
140

141
    """
142
    if req.request_method.upper() != http.HTTP_POST:
143
      raise http.HttpBadRequest("Only the POST method is supported")
144

    
145
    path = req.request_path
146
    if path.startswith("/"):
147
      path = path[1:]
148

    
149
    method = getattr(self, "perspective_%s" % path, None)
150
    if method is None:
151
      raise http.HttpNotFound()
152

    
153
    try:
154
      result = (True, method(serializer.LoadJson(req.request_body)))
155

    
156
    except backend.RPCFail, err:
157
      # our custom failure exception; str(err) works fine if the
158
      # exception was constructed with a single argument, and in
159
      # this case, err.message == err.args[0] == str(err)
160
      result = (False, str(err))
161
    except errors.QuitGanetiException, err:
162
      # Tell parent to quit
163
      logging.info("Shutting down the node daemon, arguments: %s",
164
                   str(err.args))
165
      os.kill(self.noded_pid, signal.SIGTERM)
166
      # And return the error's arguments, which must be already in
167
      # correct tuple format
168
      result = err.args
169
    except Exception, err:
170
      logging.exception("Error in RPC call")
171
      result = (False, "Error while executing backend function: %s" % str(err))
172

    
173
    return serializer.DumpJson(result)
174

    
175
  # the new block devices  --------------------------
176

    
177
  @staticmethod
178
  def perspective_blockdev_create(params):
179
    """Create a block device.
180

181
    """
182
    (bdev_s, size, owner, on_primary, info, excl_stor) = params
183
    bdev = objects.Disk.FromDict(bdev_s)
184
    if bdev is None:
185
      raise ValueError("can't unserialize data!")
186
    return backend.BlockdevCreate(bdev, size, owner, on_primary, info,
187
                                  excl_stor)
188

    
189
  @staticmethod
190
  def perspective_blockdev_pause_resume_sync(params):
191
    """Pause/resume sync of a block device.
192

193
    """
194
    disks_s, pause = params
195
    disks = [objects.Disk.FromDict(bdev_s) for bdev_s in disks_s]
196
    return backend.BlockdevPauseResumeSync(disks, pause)
197

    
198
  @staticmethod
199
  def perspective_blockdev_wipe(params):
200
    """Wipe a block device.
201

202
    """
203
    bdev_s, offset, size = params
204
    bdev = objects.Disk.FromDict(bdev_s)
205
    return backend.BlockdevWipe(bdev, offset, size)
206

    
207
  @staticmethod
208
  def perspective_blockdev_remove(params):
209
    """Remove a block device.
210

211
    """
212
    bdev_s = params[0]
213
    bdev = objects.Disk.FromDict(bdev_s)
214
    return backend.BlockdevRemove(bdev)
215

    
216
  @staticmethod
217
  def perspective_blockdev_rename(params):
218
    """Remove a block device.
219

220
    """
221
    devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params[0]]
222
    return backend.BlockdevRename(devlist)
223

    
224
  @staticmethod
225
  def perspective_blockdev_assemble(params):
226
    """Assemble a block device.
227

228
    """
229
    bdev_s, owner, on_primary, idx = params
230
    bdev = objects.Disk.FromDict(bdev_s)
231
    if bdev is None:
232
      raise ValueError("can't unserialize data!")
233
    return backend.BlockdevAssemble(bdev, owner, on_primary, idx)
234

    
235
  @staticmethod
236
  def perspective_blockdev_shutdown(params):
237
    """Shutdown a block device.
238

239
    """
240
    bdev_s = params[0]
241
    bdev = objects.Disk.FromDict(bdev_s)
242
    if bdev is None:
243
      raise ValueError("can't unserialize data!")
244
    return backend.BlockdevShutdown(bdev)
245

    
246
  @staticmethod
247
  def perspective_blockdev_addchildren(params):
248
    """Add a child to a mirror device.
249

250
    Note: this is only valid for mirror devices. It's the caller's duty
251
    to send a correct disk, otherwise we raise an error.
252

253
    """
254
    bdev_s, ndev_s = params
255
    bdev = objects.Disk.FromDict(bdev_s)
256
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
257
    if bdev is None or ndevs.count(None) > 0:
258
      raise ValueError("can't unserialize data!")
259
    return backend.BlockdevAddchildren(bdev, ndevs)
260

    
261
  @staticmethod
262
  def perspective_blockdev_removechildren(params):
263
    """Remove a child from a mirror device.
264

265
    This is only valid for mirror devices, of course. It's the callers
266
    duty to send a correct disk, otherwise we raise an error.
267

268
    """
269
    bdev_s, ndev_s = params
270
    bdev = objects.Disk.FromDict(bdev_s)
271
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
272
    if bdev is None or ndevs.count(None) > 0:
273
      raise ValueError("can't unserialize data!")
274
    return backend.BlockdevRemovechildren(bdev, ndevs)
275

    
276
  @staticmethod
277
  def perspective_blockdev_getmirrorstatus(params):
278
    """Return the mirror status for a list of disks.
279

280
    """
281
    disks = [objects.Disk.FromDict(dsk_s)
282
             for dsk_s in params[0]]
283
    return [status.ToDict()
284
            for status in backend.BlockdevGetmirrorstatus(disks)]
285

    
286
  @staticmethod
287
  def perspective_blockdev_getmirrorstatus_multi(params):
288
    """Return the mirror status for a list of disks.
289

290
    """
291
    (node_disks, ) = params
292

    
293
    disks = [objects.Disk.FromDict(dsk_s) for dsk_s in node_disks]
294

    
295
    result = []
296

    
297
    for (success, status) in backend.BlockdevGetmirrorstatusMulti(disks):
298
      if success:
299
        result.append((success, status.ToDict()))
300
      else:
301
        result.append((success, status))
302

    
303
    return result
304

    
305
  @staticmethod
306
  def perspective_blockdev_find(params):
307
    """Expose the FindBlockDevice functionality for a disk.
308

309
    This will try to find but not activate a disk.
310

311
    """
312
    disk = objects.Disk.FromDict(params[0])
313

    
314
    result = backend.BlockdevFind(disk)
315
    if result is None:
316
      return None
317

    
318
    return result.ToDict()
319

    
320
  @staticmethod
321
  def perspective_blockdev_snapshot(params):
322
    """Create a snapshot device.
323

324
    Note that this is only valid for LVM disks, if we get passed
325
    something else we raise an exception. The snapshot device can be
326
    remove by calling the generic block device remove call.
327

328
    """
329
    cfbd = objects.Disk.FromDict(params[0])
330
    return backend.BlockdevSnapshot(cfbd)
331

    
332
  @staticmethod
333
  def perspective_blockdev_grow(params):
334
    """Grow a stack of devices.
335

336
    """
337
    if len(params) < 4:
338
      raise ValueError("Received only 3 parameters in blockdev_grow,"
339
                       " old master?")
340
    cfbd = objects.Disk.FromDict(params[0])
341
    amount = params[1]
342
    dryrun = params[2]
343
    backingstore = params[3]
344
    return backend.BlockdevGrow(cfbd, amount, dryrun, backingstore)
345

    
346
  @staticmethod
347
  def perspective_blockdev_close(params):
348
    """Closes the given block devices.
349

350
    """
351
    disks = [objects.Disk.FromDict(cf) for cf in params[1]]
352
    return backend.BlockdevClose(params[0], disks)
353

    
354
  @staticmethod
355
  def perspective_blockdev_getsize(params):
356
    """Compute the sizes of the given block devices.
357

358
    """
359
    disks = [objects.Disk.FromDict(cf) for cf in params[0]]
360
    return backend.BlockdevGetsize(disks)
361

    
362
  @staticmethod
363
  def perspective_blockdev_export(params):
364
    """Compute the sizes of the given block devices.
365

366
    """
367
    disk = objects.Disk.FromDict(params[0])
368
    dest_node, dest_path, cluster_name = params[1:]
369
    return backend.BlockdevExport(disk, dest_node, dest_path, cluster_name)
370

    
371
  @staticmethod
372
  def perspective_blockdev_setinfo(params):
373
    """Sets metadata information on the given block device.
374

375
    """
376
    (disk, info) = params
377
    disk = objects.Disk.FromDict(disk)
378
    return backend.BlockdevSetInfo(disk, info)
379

    
380
  # blockdev/drbd specific methods ----------
381

    
382
  @staticmethod
383
  def perspective_drbd_disconnect_net(params):
384
    """Disconnects the network connection of drbd disks.
385

386
    Note that this is only valid for drbd disks, so the members of the
387
    disk list must all be drbd devices.
388

389
    """
390
    nodes_ip, disks = params
391
    disks = [objects.Disk.FromDict(cf) for cf in disks]
392
    return backend.DrbdDisconnectNet(nodes_ip, disks)
393

    
394
  @staticmethod
395
  def perspective_drbd_attach_net(params):
396
    """Attaches the network connection of drbd disks.
397

398
    Note that this is only valid for drbd disks, so the members of the
399
    disk list must all be drbd devices.
400

401
    """
402
    nodes_ip, disks, instance_name, multimaster = params
403
    disks = [objects.Disk.FromDict(cf) for cf in disks]
404
    return backend.DrbdAttachNet(nodes_ip, disks,
405
                                     instance_name, multimaster)
406

    
407
  @staticmethod
408
  def perspective_drbd_wait_sync(params):
409
    """Wait until DRBD disks are synched.
410

411
    Note that this is only valid for drbd disks, so the members of the
412
    disk list must all be drbd devices.
413

414
    """
415
    nodes_ip, disks = params
416
    disks = [objects.Disk.FromDict(cf) for cf in disks]
417
    return backend.DrbdWaitSync(nodes_ip, disks)
418

    
419
  @staticmethod
420
  def perspective_drbd_helper(params):
421
    """Query drbd helper.
422

423
    """
424
    return backend.GetDrbdUsermodeHelper()
425

    
426
  # export/import  --------------------------
427

    
428
  @staticmethod
429
  def perspective_finalize_export(params):
430
    """Expose the finalize export functionality.
431

432
    """
433
    instance = objects.Instance.FromDict(params[0])
434

    
435
    snap_disks = []
436
    for disk in params[1]:
437
      if isinstance(disk, bool):
438
        snap_disks.append(disk)
439
      else:
440
        snap_disks.append(objects.Disk.FromDict(disk))
441

    
442
    return backend.FinalizeExport(instance, snap_disks)
443

    
444
  @staticmethod
445
  def perspective_export_info(params):
446
    """Query information about an existing export on this node.
447

448
    The given path may not contain an export, in which case we return
449
    None.
450

451
    """
452
    path = params[0]
453
    return backend.ExportInfo(path)
454

    
455
  @staticmethod
456
  def perspective_export_list(params):
457
    """List the available exports on this node.
458

459
    Note that as opposed to export_info, which may query data about an
460
    export in any path, this only queries the standard Ganeti path
461
    (pathutils.EXPORT_DIR).
462

463
    """
464
    return backend.ListExports()
465

    
466
  @staticmethod
467
  def perspective_export_remove(params):
468
    """Remove an export.
469

470
    """
471
    export = params[0]
472
    return backend.RemoveExport(export)
473

    
474
  # block device ---------------------
475
  @staticmethod
476
  def perspective_bdev_sizes(params):
477
    """Query the list of block devices
478

479
    """
480
    devices = params[0]
481
    return backend.GetBlockDevSizes(devices)
482

    
483
  # volume  --------------------------
484

    
485
  @staticmethod
486
  def perspective_lv_list(params):
487
    """Query the list of logical volumes in a given volume group.
488

489
    """
490
    vgname = params[0]
491
    return backend.GetVolumeList(vgname)
492

    
493
  @staticmethod
494
  def perspective_vg_list(params):
495
    """Query the list of volume groups.
496

497
    """
498
    return backend.ListVolumeGroups()
499

    
500
  # Storage --------------------------
501

    
502
  @staticmethod
503
  def perspective_storage_list(params):
504
    """Get list of storage units.
505

506
    """
507
    (su_name, su_args, name, fields) = params
508
    return storage.GetStorage(su_name, *su_args).List(name, fields)
509

    
510
  @staticmethod
511
  def perspective_storage_modify(params):
512
    """Modify a storage unit.
513

514
    """
515
    (su_name, su_args, name, changes) = params
516
    return storage.GetStorage(su_name, *su_args).Modify(name, changes)
517

    
518
  @staticmethod
519
  def perspective_storage_execute(params):
520
    """Execute an operation on a storage unit.
521

522
    """
523
    (su_name, su_args, name, op) = params
524
    return storage.GetStorage(su_name, *su_args).Execute(name, op)
525

    
526
  # bridge  --------------------------
527

    
528
  @staticmethod
529
  def perspective_bridges_exist(params):
530
    """Check if all bridges given exist on this node.
531

532
    """
533
    bridges_list = params[0]
534
    return backend.BridgesExist(bridges_list)
535

    
536
  # instance  --------------------------
537

    
538
  @staticmethod
539
  def perspective_instance_os_add(params):
540
    """Install an OS on a given instance.
541

542
    """
543
    inst_s = params[0]
544
    inst = objects.Instance.FromDict(inst_s)
545
    reinstall = params[1]
546
    debug = params[2]
547
    return backend.InstanceOsAdd(inst, reinstall, debug)
548

    
549
  @staticmethod
550
  def perspective_instance_run_rename(params):
551
    """Runs the OS rename script for an instance.
552

553
    """
554
    inst_s, old_name, debug = params
555
    inst = objects.Instance.FromDict(inst_s)
556
    return backend.RunRenameInstance(inst, old_name, debug)
557

    
558
  @staticmethod
559
  def perspective_instance_shutdown(params):
560
    """Shutdown an instance.
561

562
    """
563
    instance = objects.Instance.FromDict(params[0])
564
    timeout = params[1]
565
    return backend.InstanceShutdown(instance, timeout)
566

    
567
  @staticmethod
568
  def perspective_instance_start(params):
569
    """Start an instance.
570

571
    """
572
    (instance_name, startup_paused) = params
573
    instance = objects.Instance.FromDict(instance_name)
574
    return backend.StartInstance(instance, startup_paused)
575

    
576
  @staticmethod
577
  def perspective_hot_add_disk(params):
578
    """Hotplugs a nic to a running instance.
579

580
    """
581
    (idict, ddict, dev_path, seq) = params
582
    instance = objects.Instance.FromDict(idict)
583
    disk = objects.Disk.FromDict(ddict)
584
    return backend.HotAddDisk(instance, disk, dev_path, seq)
585

    
586
  @staticmethod
587
  def perspective_hot_del_disk(params):
588
    """Hotplugs a nic to a running instance.
589

590
    """
591
    (idict, ddict, seq) = params
592
    instance = objects.Instance.FromDict(idict)
593
    disk = objects.Disk.FromDict(ddict)
594
    return backend.HotDelDisk(instance, disk, seq)
595

    
596
  @staticmethod
597
  def perspective_hot_add_nic(params):
598
    """Hotplugs a nic to a running instance.
599

600
    """
601
    (idict, ndict, seq) = params
602
    instance = objects.Instance.FromDict(idict)
603
    nic = objects.NIC.FromDict(ndict)
604
    return backend.HotAddNic(instance, nic, seq)
605

    
606
  @staticmethod
607
  def perspective_hot_del_nic(params):
608
    """Hotplugs a nic to a running instance.
609

610
    """
611
    (idict, ndict, seq) = params
612
    instance = objects.Instance.FromDict(idict)
613
    nic = objects.NIC.FromDict(ndict)
614
    return backend.HotDelNic(instance, nic, seq)
615

    
616
  @staticmethod
617
  def perspective_migration_info(params):
618
    """Gather information about an instance to be migrated.
619

620
    """
621
    instance = objects.Instance.FromDict(params[0])
622
    return backend.MigrationInfo(instance)
623

    
624
  @staticmethod
625
  def perspective_accept_instance(params):
626
    """Prepare the node to accept an instance.
627

628
    """
629
    instance, info, target = params
630
    instance = objects.Instance.FromDict(instance)
631
    return backend.AcceptInstance(instance, info, target)
632

    
633
  @staticmethod
634
  def perspective_instance_finalize_migration_dst(params):
635
    """Finalize the instance migration on the destination node.
636

637
    """
638
    instance, info, success = params
639
    instance = objects.Instance.FromDict(instance)
640
    return backend.FinalizeMigrationDst(instance, info, success)
641

    
642
  @staticmethod
643
  def perspective_instance_migrate(params):
644
    """Migrates an instance.
645

646
    """
647
    instance, target, live = params
648
    instance = objects.Instance.FromDict(instance)
649
    return backend.MigrateInstance(instance, target, live)
650

    
651
  @staticmethod
652
  def perspective_instance_finalize_migration_src(params):
653
    """Finalize the instance migration on the source node.
654

655
    """
656
    instance, success, live = params
657
    instance = objects.Instance.FromDict(instance)
658
    return backend.FinalizeMigrationSource(instance, success, live)
659

    
660
  @staticmethod
661
  def perspective_instance_get_migration_status(params):
662
    """Reports migration status.
663

664
    """
665
    instance = objects.Instance.FromDict(params[0])
666
    return backend.GetMigrationStatus(instance).ToDict()
667

    
668
  @staticmethod
669
  def perspective_instance_reboot(params):
670
    """Reboot an instance.
671

672
    """
673
    instance = objects.Instance.FromDict(params[0])
674
    reboot_type = params[1]
675
    shutdown_timeout = params[2]
676
    return backend.InstanceReboot(instance, reboot_type, shutdown_timeout)
677

    
678
  @staticmethod
679
  def perspective_instance_balloon_memory(params):
680
    """Modify instance runtime memory.
681

682
    """
683
    instance_dict, memory = params
684
    instance = objects.Instance.FromDict(instance_dict)
685
    return backend.InstanceBalloonMemory(instance, memory)
686

    
687
  @staticmethod
688
  def perspective_instance_info(params):
689
    """Query instance information.
690

691
    """
692
    return backend.GetInstanceInfo(params[0], params[1])
693

    
694
  @staticmethod
695
  def perspective_instance_migratable(params):
696
    """Query whether the specified instance can be migrated.
697

698
    """
699
    instance = objects.Instance.FromDict(params[0])
700
    return backend.GetInstanceMigratable(instance)
701

    
702
  @staticmethod
703
  def perspective_all_instances_info(params):
704
    """Query information about all instances.
705

706
    """
707
    return backend.GetAllInstancesInfo(params[0])
708

    
709
  @staticmethod
710
  def perspective_instance_list(params):
711
    """Query the list of running instances.
712

713
    """
714
    return backend.GetInstanceList(params[0])
715

    
716
  # node --------------------------
717

    
718
  @staticmethod
719
  def perspective_node_has_ip_address(params):
720
    """Checks if a node has the given ip address.
721

722
    """
723
    return netutils.IPAddress.Own(params[0])
724

    
725
  @staticmethod
726
  def perspective_node_info(params):
727
    """Query node information.
728

729
    """
730
    (vg_names, hv_names, excl_stor) = params
731
    return backend.GetNodeInfo(vg_names, hv_names, excl_stor)
732

    
733
  @staticmethod
734
  def perspective_etc_hosts_modify(params):
735
    """Modify a node entry in /etc/hosts.
736

737
    """
738
    backend.EtcHostsModify(params[0], params[1], params[2])
739

    
740
    return True
741

    
742
  @staticmethod
743
  def perspective_node_verify(params):
744
    """Run a verify sequence on this node.
745

746
    """
747
    return backend.VerifyNode(params[0], params[1])
748

    
749
  @classmethod
750
  def perspective_node_verify_light(cls, params):
751
    """Run a light verify sequence on this node.
752

753
    """
754
    # So far it's the same as the normal node_verify
755
    return cls.perspective_node_verify(params)
756

    
757
  @staticmethod
758
  def perspective_node_start_master_daemons(params):
759
    """Start the master daemons on this node.
760

761
    """
762
    return backend.StartMasterDaemons(params[0])
763

    
764
  @staticmethod
765
  def perspective_node_activate_master_ip(params):
766
    """Activate the master IP on this node.
767

768
    """
769
    master_params = objects.MasterNetworkParameters.FromDict(params[0])
770
    return backend.ActivateMasterIp(master_params, params[1])
771

    
772
  @staticmethod
773
  def perspective_node_deactivate_master_ip(params):
774
    """Deactivate the master IP on this node.
775

776
    """
777
    master_params = objects.MasterNetworkParameters.FromDict(params[0])
778
    return backend.DeactivateMasterIp(master_params, params[1])
779

    
780
  @staticmethod
781
  def perspective_node_stop_master(params):
782
    """Stops master daemons on this node.
783

784
    """
785
    return backend.StopMasterDaemons()
786

    
787
  @staticmethod
788
  def perspective_node_change_master_netmask(params):
789
    """Change the master IP netmask.
790

791
    """
792
    return backend.ChangeMasterNetmask(params[0], params[1], params[2],
793
                                       params[3])
794

    
795
  @staticmethod
796
  def perspective_node_leave_cluster(params):
797
    """Cleanup after leaving a cluster.
798

799
    """
800
    return backend.LeaveCluster(params[0])
801

    
802
  @staticmethod
803
  def perspective_node_volumes(params):
804
    """Query the list of all logical volume groups.
805

806
    """
807
    return backend.NodeVolumes()
808

    
809
  @staticmethod
810
  def perspective_node_demote_from_mc(params):
811
    """Demote a node from the master candidate role.
812

813
    """
814
    return backend.DemoteFromMC()
815

    
816
  @staticmethod
817
  def perspective_node_powercycle(params):
818
    """Tries to powercycle the nod.
819

820
    """
821
    hypervisor_type = params[0]
822
    return backend.PowercycleNode(hypervisor_type)
823

    
824
  # cluster --------------------------
825

    
826
  @staticmethod
827
  def perspective_version(params):
828
    """Query version information.
829

830
    """
831
    return constants.PROTOCOL_VERSION
832

    
833
  @staticmethod
834
  def perspective_upload_file(params):
835
    """Upload a file.
836

837
    Note that the backend implementation imposes strict rules on which
838
    files are accepted.
839

840
    """
841
    return backend.UploadFile(*(params[0]))
842

    
843
  @staticmethod
844
  def perspective_master_info(params):
845
    """Query master information.
846

847
    """
848
    return backend.GetMasterInfo()
849

    
850
  @staticmethod
851
  def perspective_run_oob(params):
852
    """Runs oob on node.
853

854
    """
855
    output = backend.RunOob(params[0], params[1], params[2], params[3])
856
    if output:
857
      result = serializer.LoadJson(output)
858
    else:
859
      result = None
860
    return result
861

    
862
  @staticmethod
863
  def perspective_restricted_command(params):
864
    """Runs a restricted command.
865

866
    """
867
    (cmd, ) = params
868

    
869
    return backend.RunRestrictedCmd(cmd)
870

    
871
  @staticmethod
872
  def perspective_write_ssconf_files(params):
873
    """Write ssconf files.
874

875
    """
876
    (values,) = params
877
    return ssconf.WriteSsconfFiles(values)
878

    
879
  @staticmethod
880
  def perspective_get_watcher_pause(params):
881
    """Get watcher pause end.
882

883
    """
884
    return utils.ReadWatcherPauseFile(pathutils.WATCHER_PAUSEFILE)
885

    
886
  @staticmethod
887
  def perspective_set_watcher_pause(params):
888
    """Set watcher pause.
889

890
    """
891
    (until, ) = params
892
    return backend.SetWatcherPause(until)
893

    
894
  # os -----------------------
895

    
896
  @staticmethod
897
  def perspective_os_diagnose(params):
898
    """Query detailed information about existing OSes.
899

900
    """
901
    return backend.DiagnoseOS()
902

    
903
  @staticmethod
904
  def perspective_os_get(params):
905
    """Query information about a given OS.
906

907
    """
908
    name = params[0]
909
    os_obj = backend.OSFromDisk(name)
910
    return os_obj.ToDict()
911

    
912
  @staticmethod
913
  def perspective_os_validate(params):
914
    """Run a given OS' validation routine.
915

916
    """
917
    required, name, checks, params = params
918
    return backend.ValidateOS(required, name, checks, params)
919

    
920
  # extstorage -----------------------
921

    
922
  @staticmethod
923
  def perspective_extstorage_diagnose(params):
924
    """Query detailed information about existing extstorage providers.
925

926
    """
927
    return backend.DiagnoseExtStorage()
928

    
929
  # hooks -----------------------
930

    
931
  @staticmethod
932
  def perspective_hooks_runner(params):
933
    """Run hook scripts.
934

935
    """
936
    hpath, phase, env = params
937
    hr = backend.HooksRunner()
938
    return hr.RunHooks(hpath, phase, env)
939

    
940
  # iallocator -----------------
941

    
942
  @staticmethod
943
  def perspective_iallocator_runner(params):
944
    """Run an iallocator script.
945

946
    """
947
    name, idata = params
948
    iar = backend.IAllocatorRunner()
949
    return iar.Run(name, idata)
950

    
951
  # test -----------------------
952

    
953
  @staticmethod
954
  def perspective_test_delay(params):
955
    """Run test delay.
956

957
    """
958
    duration = params[0]
959
    status, rval = utils.TestDelay(duration)
960
    if not status:
961
      raise backend.RPCFail(rval)
962
    return rval
963

    
964
  # file storage ---------------
965

    
966
  @staticmethod
967
  def perspective_file_storage_dir_create(params):
968
    """Create the file storage directory.
969

970
    """
971
    file_storage_dir = params[0]
972
    return backend.CreateFileStorageDir(file_storage_dir)
973

    
974
  @staticmethod
975
  def perspective_file_storage_dir_remove(params):
976
    """Remove the file storage directory.
977

978
    """
979
    file_storage_dir = params[0]
980
    return backend.RemoveFileStorageDir(file_storage_dir)
981

    
982
  @staticmethod
983
  def perspective_file_storage_dir_rename(params):
984
    """Rename the file storage directory.
985

986
    """
987
    old_file_storage_dir = params[0]
988
    new_file_storage_dir = params[1]
989
    return backend.RenameFileStorageDir(old_file_storage_dir,
990
                                        new_file_storage_dir)
991

    
992
  # jobs ------------------------
993

    
994
  @staticmethod
995
  @_RequireJobQueueLock
996
  def perspective_jobqueue_update(params):
997
    """Update job queue.
998

999
    """
1000
    (file_name, content) = params
1001
    return backend.JobQueueUpdate(file_name, content)
1002

    
1003
  @staticmethod
1004
  @_RequireJobQueueLock
1005
  def perspective_jobqueue_purge(params):
1006
    """Purge job queue.
1007

1008
    """
1009
    return backend.JobQueuePurge()
1010

    
1011
  @staticmethod
1012
  @_RequireJobQueueLock
1013
  def perspective_jobqueue_rename(params):
1014
    """Rename a job queue file.
1015

1016
    """
1017
    # TODO: What if a file fails to rename?
1018
    return [backend.JobQueueRename(old, new) for old, new in params[0]]
1019

    
1020
  @staticmethod
1021
  @_RequireJobQueueLock
1022
  def perspective_jobqueue_set_drain_flag(params):
1023
    """Set job queue's drain flag.
1024

1025
    """
1026
    (flag, ) = params
1027

    
1028
    return jstore.SetDrainFlag(flag)
1029

    
1030
  # hypervisor ---------------
1031

    
1032
  @staticmethod
1033
  def perspective_hypervisor_validate_params(params):
1034
    """Validate the hypervisor parameters.
1035

1036
    """
1037
    (hvname, hvparams) = params
1038
    return backend.ValidateHVParams(hvname, hvparams)
1039

    
1040
  # Crypto
1041

    
1042
  @staticmethod
1043
  def perspective_x509_cert_create(params):
1044
    """Creates a new X509 certificate for SSL/TLS.
1045

1046
    """
1047
    (validity, ) = params
1048
    return backend.CreateX509Certificate(validity)
1049

    
1050
  @staticmethod
1051
  def perspective_x509_cert_remove(params):
1052
    """Removes a X509 certificate.
1053

1054
    """
1055
    (name, ) = params
1056
    return backend.RemoveX509Certificate(name)
1057

    
1058
  # Import and export
1059

    
1060
  @staticmethod
1061
  def perspective_import_start(params):
1062
    """Starts an import daemon.
1063

1064
    """
1065
    (opts_s, instance, component, (dest, dest_args)) = params
1066

    
1067
    opts = objects.ImportExportOptions.FromDict(opts_s)
1068

    
1069
    return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
1070
                                           None, None,
1071
                                           objects.Instance.FromDict(instance),
1072
                                           component, dest,
1073
                                           _DecodeImportExportIO(dest,
1074
                                                                 dest_args))
1075

    
1076
  @staticmethod
1077
  def perspective_export_start(params):
1078
    """Starts an export daemon.
1079

1080
    """
1081
    (opts_s, host, port, instance, component, (source, source_args)) = params
1082

    
1083
    opts = objects.ImportExportOptions.FromDict(opts_s)
1084

    
1085
    return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
1086
                                           host, port,
1087
                                           objects.Instance.FromDict(instance),
1088
                                           component, source,
1089
                                           _DecodeImportExportIO(source,
1090
                                                                 source_args))
1091

    
1092
  @staticmethod
1093
  def perspective_impexp_status(params):
1094
    """Retrieves the status of an import or export daemon.
1095

1096
    """
1097
    return backend.GetImportExportStatus(params[0])
1098

    
1099
  @staticmethod
1100
  def perspective_impexp_abort(params):
1101
    """Aborts an import or export.
1102

1103
    """
1104
    return backend.AbortImportExport(params[0])
1105

    
1106
  @staticmethod
1107
  def perspective_impexp_cleanup(params):
1108
    """Cleans up after an import or export.
1109

1110
    """
1111
    return backend.CleanupImportExport(params[0])
1112

    
1113

    
1114
def CheckNoded(_, args):
1115
  """Initial checks whether to run or exit with a failure.
1116

1117
  """
1118
  if args: # noded doesn't take any arguments
1119
    print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
1120
                          sys.argv[0])
1121
    sys.exit(constants.EXIT_FAILURE)
1122
  try:
1123
    codecs.lookup("string-escape")
1124
  except LookupError:
1125
    print >> sys.stderr, ("Can't load the string-escape code which is part"
1126
                          " of the Python installation. Is your installation"
1127
                          " complete/correct? Aborting.")
1128
    sys.exit(constants.EXIT_FAILURE)
1129

    
1130

    
1131
def PrepNoded(options, _):
1132
  """Preparation node daemon function, executed with the PID file held.
1133

1134
  """
1135
  if options.mlock:
1136
    request_executor_class = MlockallRequestExecutor
1137
    try:
1138
      utils.Mlockall()
1139
    except errors.NoCtypesError:
1140
      logging.warning("Cannot set memory lock, ctypes module not found")
1141
      request_executor_class = http.server.HttpServerRequestExecutor
1142
  else:
1143
    request_executor_class = http.server.HttpServerRequestExecutor
1144

    
1145
  # Read SSL certificate
1146
  if options.ssl:
1147
    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
1148
                                    ssl_cert_path=options.ssl_cert)
1149
  else:
1150
    ssl_params = None
1151

    
1152
  err = _PrepareQueueLock()
1153
  if err is not None:
1154
    # this might be some kind of file-system/permission error; while
1155
    # this breaks the job queue functionality, we shouldn't prevent
1156
    # startup of the whole node daemon because of this
1157
    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
1158

    
1159
  handler = NodeRequestHandler()
1160

    
1161
  mainloop = daemon.Mainloop()
1162
  server = \
1163
    http.server.HttpServer(mainloop, options.bind_address, options.port,
1164
                           handler, ssl_params=ssl_params, ssl_verify_peer=True,
1165
                           request_executor_class=request_executor_class)
1166
  server.Start()
1167

    
1168
  return (mainloop, server)
1169

    
1170

    
1171
def ExecNoded(options, args, prep_data): # pylint: disable=W0613
1172
  """Main node daemon function, executed with the PID file held.
1173

1174
  """
1175
  (mainloop, server) = prep_data
1176
  try:
1177
    mainloop.Run()
1178
  finally:
1179
    server.Stop()
1180

    
1181

    
1182
def Main():
1183
  """Main function for the node daemon.
1184

1185
  """
1186
  parser = OptionParser(description="Ganeti node daemon",
1187
                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]",
1188
                        version="%%prog (ganeti) %s" %
1189
                        constants.RELEASE_VERSION)
1190
  parser.add_option("--no-mlock", dest="mlock",
1191
                    help="Do not mlock the node memory in ram",
1192
                    default=True, action="store_false")
1193

    
1194
  daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded,
1195
                     default_ssl_cert=pathutils.NODED_CERT_FILE,
1196
                     default_ssl_key=pathutils.NODED_CERT_FILE,
1197
                     console_logging=True)