Statistics
| Branch: | Tag: | Revision:

root / lib / server / noded.py @ bbc4783a

History | View | Annotate | Download (31.5 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
  @staticmethod
750
  def perspective_node_start_master_daemons(params):
751
    """Start the master daemons on this node.
752

753
    """
754
    return backend.StartMasterDaemons(params[0])
755

    
756
  @staticmethod
757
  def perspective_node_activate_master_ip(params):
758
    """Activate the master IP on this node.
759

760
    """
761
    master_params = objects.MasterNetworkParameters.FromDict(params[0])
762
    return backend.ActivateMasterIp(master_params, params[1])
763

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

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

    
772
  @staticmethod
773
  def perspective_node_stop_master(params):
774
    """Stops master daemons on this node.
775

776
    """
777
    return backend.StopMasterDaemons()
778

    
779
  @staticmethod
780
  def perspective_node_change_master_netmask(params):
781
    """Change the master IP netmask.
782

783
    """
784
    return backend.ChangeMasterNetmask(params[0], params[1], params[2],
785
                                       params[3])
786

    
787
  @staticmethod
788
  def perspective_node_leave_cluster(params):
789
    """Cleanup after leaving a cluster.
790

791
    """
792
    return backend.LeaveCluster(params[0])
793

    
794
  @staticmethod
795
  def perspective_node_volumes(params):
796
    """Query the list of all logical volume groups.
797

798
    """
799
    return backend.NodeVolumes()
800

    
801
  @staticmethod
802
  def perspective_node_demote_from_mc(params):
803
    """Demote a node from the master candidate role.
804

805
    """
806
    return backend.DemoteFromMC()
807

    
808
  @staticmethod
809
  def perspective_node_powercycle(params):
810
    """Tries to powercycle the nod.
811

812
    """
813
    hypervisor_type = params[0]
814
    return backend.PowercycleNode(hypervisor_type)
815

    
816
  # cluster --------------------------
817

    
818
  @staticmethod
819
  def perspective_version(params):
820
    """Query version information.
821

822
    """
823
    return constants.PROTOCOL_VERSION
824

    
825
  @staticmethod
826
  def perspective_upload_file(params):
827
    """Upload a file.
828

829
    Note that the backend implementation imposes strict rules on which
830
    files are accepted.
831

832
    """
833
    return backend.UploadFile(*(params[0]))
834

    
835
  @staticmethod
836
  def perspective_master_info(params):
837
    """Query master information.
838

839
    """
840
    return backend.GetMasterInfo()
841

    
842
  @staticmethod
843
  def perspective_run_oob(params):
844
    """Runs oob on node.
845

846
    """
847
    output = backend.RunOob(params[0], params[1], params[2], params[3])
848
    if output:
849
      result = serializer.LoadJson(output)
850
    else:
851
      result = None
852
    return result
853

    
854
  @staticmethod
855
  def perspective_restricted_command(params):
856
    """Runs a restricted command.
857

858
    """
859
    (cmd, ) = params
860

    
861
    return backend.RunRestrictedCmd(cmd)
862

    
863
  @staticmethod
864
  def perspective_write_ssconf_files(params):
865
    """Write ssconf files.
866

867
    """
868
    (values,) = params
869
    return ssconf.WriteSsconfFiles(values)
870

    
871
  @staticmethod
872
  def perspective_get_watcher_pause(params):
873
    """Get watcher pause end.
874

875
    """
876
    return utils.ReadWatcherPauseFile(pathutils.WATCHER_PAUSEFILE)
877

    
878
  @staticmethod
879
  def perspective_set_watcher_pause(params):
880
    """Set watcher pause.
881

882
    """
883
    (until, ) = params
884
    return backend.SetWatcherPause(until)
885

    
886
  # os -----------------------
887

    
888
  @staticmethod
889
  def perspective_os_diagnose(params):
890
    """Query detailed information about existing OSes.
891

892
    """
893
    return backend.DiagnoseOS()
894

    
895
  @staticmethod
896
  def perspective_os_get(params):
897
    """Query information about a given OS.
898

899
    """
900
    name = params[0]
901
    os_obj = backend.OSFromDisk(name)
902
    return os_obj.ToDict()
903

    
904
  @staticmethod
905
  def perspective_os_validate(params):
906
    """Run a given OS' validation routine.
907

908
    """
909
    required, name, checks, params = params
910
    return backend.ValidateOS(required, name, checks, params)
911

    
912
  # extstorage -----------------------
913

    
914
  @staticmethod
915
  def perspective_extstorage_diagnose(params):
916
    """Query detailed information about existing extstorage providers.
917

918
    """
919
    return backend.DiagnoseExtStorage()
920

    
921
  # hooks -----------------------
922

    
923
  @staticmethod
924
  def perspective_hooks_runner(params):
925
    """Run hook scripts.
926

927
    """
928
    hpath, phase, env = params
929
    hr = backend.HooksRunner()
930
    return hr.RunHooks(hpath, phase, env)
931

    
932
  # iallocator -----------------
933

    
934
  @staticmethod
935
  def perspective_iallocator_runner(params):
936
    """Run an iallocator script.
937

938
    """
939
    name, idata = params
940
    iar = backend.IAllocatorRunner()
941
    return iar.Run(name, idata)
942

    
943
  # test -----------------------
944

    
945
  @staticmethod
946
  def perspective_test_delay(params):
947
    """Run test delay.
948

949
    """
950
    duration = params[0]
951
    status, rval = utils.TestDelay(duration)
952
    if not status:
953
      raise backend.RPCFail(rval)
954
    return rval
955

    
956
  # file storage ---------------
957

    
958
  @staticmethod
959
  def perspective_file_storage_dir_create(params):
960
    """Create the file storage directory.
961

962
    """
963
    file_storage_dir = params[0]
964
    return backend.CreateFileStorageDir(file_storage_dir)
965

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

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

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

978
    """
979
    old_file_storage_dir = params[0]
980
    new_file_storage_dir = params[1]
981
    return backend.RenameFileStorageDir(old_file_storage_dir,
982
                                        new_file_storage_dir)
983

    
984
  # jobs ------------------------
985

    
986
  @staticmethod
987
  @_RequireJobQueueLock
988
  def perspective_jobqueue_update(params):
989
    """Update job queue.
990

991
    """
992
    (file_name, content) = params
993
    return backend.JobQueueUpdate(file_name, content)
994

    
995
  @staticmethod
996
  @_RequireJobQueueLock
997
  def perspective_jobqueue_purge(params):
998
    """Purge job queue.
999

1000
    """
1001
    return backend.JobQueuePurge()
1002

    
1003
  @staticmethod
1004
  @_RequireJobQueueLock
1005
  def perspective_jobqueue_rename(params):
1006
    """Rename a job queue file.
1007

1008
    """
1009
    # TODO: What if a file fails to rename?
1010
    return [backend.JobQueueRename(old, new) for old, new in params[0]]
1011

    
1012
  @staticmethod
1013
  @_RequireJobQueueLock
1014
  def perspective_jobqueue_set_drain_flag(params):
1015
    """Set job queue's drain flag.
1016

1017
    """
1018
    (flag, ) = params
1019

    
1020
    return jstore.SetDrainFlag(flag)
1021

    
1022
  # hypervisor ---------------
1023

    
1024
  @staticmethod
1025
  def perspective_hypervisor_validate_params(params):
1026
    """Validate the hypervisor parameters.
1027

1028
    """
1029
    (hvname, hvparams) = params
1030
    return backend.ValidateHVParams(hvname, hvparams)
1031

    
1032
  # Crypto
1033

    
1034
  @staticmethod
1035
  def perspective_x509_cert_create(params):
1036
    """Creates a new X509 certificate for SSL/TLS.
1037

1038
    """
1039
    (validity, ) = params
1040
    return backend.CreateX509Certificate(validity)
1041

    
1042
  @staticmethod
1043
  def perspective_x509_cert_remove(params):
1044
    """Removes a X509 certificate.
1045

1046
    """
1047
    (name, ) = params
1048
    return backend.RemoveX509Certificate(name)
1049

    
1050
  # Import and export
1051

    
1052
  @staticmethod
1053
  def perspective_import_start(params):
1054
    """Starts an import daemon.
1055

1056
    """
1057
    (opts_s, instance, component, (dest, dest_args)) = params
1058

    
1059
    opts = objects.ImportExportOptions.FromDict(opts_s)
1060

    
1061
    return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
1062
                                           None, None,
1063
                                           objects.Instance.FromDict(instance),
1064
                                           component, dest,
1065
                                           _DecodeImportExportIO(dest,
1066
                                                                 dest_args))
1067

    
1068
  @staticmethod
1069
  def perspective_export_start(params):
1070
    """Starts an export daemon.
1071

1072
    """
1073
    (opts_s, host, port, instance, component, (source, source_args)) = params
1074

    
1075
    opts = objects.ImportExportOptions.FromDict(opts_s)
1076

    
1077
    return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
1078
                                           host, port,
1079
                                           objects.Instance.FromDict(instance),
1080
                                           component, source,
1081
                                           _DecodeImportExportIO(source,
1082
                                                                 source_args))
1083

    
1084
  @staticmethod
1085
  def perspective_impexp_status(params):
1086
    """Retrieves the status of an import or export daemon.
1087

1088
    """
1089
    return backend.GetImportExportStatus(params[0])
1090

    
1091
  @staticmethod
1092
  def perspective_impexp_abort(params):
1093
    """Aborts an import or export.
1094

1095
    """
1096
    return backend.AbortImportExport(params[0])
1097

    
1098
  @staticmethod
1099
  def perspective_impexp_cleanup(params):
1100
    """Cleans up after an import or export.
1101

1102
    """
1103
    return backend.CleanupImportExport(params[0])
1104

    
1105

    
1106
def CheckNoded(_, args):
1107
  """Initial checks whether to run or exit with a failure.
1108

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

    
1122

    
1123
def PrepNoded(options, _):
1124
  """Preparation node daemon function, executed with the PID file held.
1125

1126
  """
1127
  if options.mlock:
1128
    request_executor_class = MlockallRequestExecutor
1129
    try:
1130
      utils.Mlockall()
1131
    except errors.NoCtypesError:
1132
      logging.warning("Cannot set memory lock, ctypes module not found")
1133
      request_executor_class = http.server.HttpServerRequestExecutor
1134
  else:
1135
    request_executor_class = http.server.HttpServerRequestExecutor
1136

    
1137
  # Read SSL certificate
1138
  if options.ssl:
1139
    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
1140
                                    ssl_cert_path=options.ssl_cert)
1141
  else:
1142
    ssl_params = None
1143

    
1144
  err = _PrepareQueueLock()
1145
  if err is not None:
1146
    # this might be some kind of file-system/permission error; while
1147
    # this breaks the job queue functionality, we shouldn't prevent
1148
    # startup of the whole node daemon because of this
1149
    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
1150

    
1151
  handler = NodeRequestHandler()
1152

    
1153
  mainloop = daemon.Mainloop()
1154
  server = \
1155
    http.server.HttpServer(mainloop, options.bind_address, options.port,
1156
                           handler, ssl_params=ssl_params, ssl_verify_peer=True,
1157
                           request_executor_class=request_executor_class)
1158
  server.Start()
1159

    
1160
  return (mainloop, server)
1161

    
1162

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

1166
  """
1167
  (mainloop, server) = prep_data
1168
  try:
1169
    mainloop.Run()
1170
  finally:
1171
    server.Stop()
1172

    
1173

    
1174
def Main():
1175
  """Main function for the node daemon.
1176

1177
  """
1178
  parser = OptionParser(description="Ganeti node daemon",
1179
                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]",
1180
                        version="%%prog (ganeti) %s" %
1181
                        constants.RELEASE_VERSION)
1182
  parser.add_option("--no-mlock", dest="mlock",
1183
                    help="Do not mlock the node memory in ram",
1184
                    default=True, action="store_false")
1185

    
1186
  daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded,
1187
                     default_ssl_cert=pathutils.NODED_CERT_FILE,
1188
                     default_ssl_key=pathutils.NODED_CERT_FILE,
1189
                     console_logging=True)