Statistics
| Branch: | Tag: | Revision:

root / lib / server / noded.py @ 90d8d4d1

History | View | Annotate | Download (32.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.storage import container
49
from ganeti import serializer
50
from ganeti import netutils
51
from ganeti import pathutils
52
from ganeti import ssconf
53

    
54
import ganeti.http.server # pylint: disable=W0611
55

    
56

    
57
queue_lock = None
58

    
59

    
60
def _extendReasonTrail(trail, source, reason=""):
61
  """Extend the reason trail with noded information
62

63
  The trail is extended by appending the name of the noded functionality
64
  """
65
  assert trail is not None
66
  trail_source = "%s:%s" % (constants.OPCODE_REASON_SRC_NODED, source)
67
  trail.append((trail_source, reason, utils.EpochNano()))
68

    
69

    
70
def _PrepareQueueLock():
71
  """Try to prepare the queue lock.
72

73
  @return: None for success, otherwise an exception object
74

75
  """
76
  global queue_lock # pylint: disable=W0603
77

    
78
  if queue_lock is not None:
79
    return None
80

    
81
  # Prepare job queue
82
  try:
83
    queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
84
    return None
85
  except EnvironmentError, err:
86
    return err
87

    
88

    
89
def _RequireJobQueueLock(fn):
90
  """Decorator for job queue manipulating functions.
91

92
  """
93
  QUEUE_LOCK_TIMEOUT = 10
94

    
95
  def wrapper(*args, **kwargs):
96
    # Locking in exclusive, blocking mode because there could be several
97
    # children running at the same time. Waiting up to 10 seconds.
98
    if _PrepareQueueLock() is not None:
99
      raise errors.JobQueueError("Job queue failed initialization,"
100
                                 " cannot update jobs")
101
    queue_lock.Exclusive(blocking=True, timeout=QUEUE_LOCK_TIMEOUT)
102
    try:
103
      return fn(*args, **kwargs)
104
    finally:
105
      queue_lock.Unlock()
106

    
107
  return wrapper
108

    
109

    
110
def _DecodeImportExportIO(ieio, ieioargs):
111
  """Decodes import/export I/O information.
112

113
  """
114
  if ieio == constants.IEIO_RAW_DISK:
115
    assert len(ieioargs) == 1
116
    return (objects.Disk.FromDict(ieioargs[0]), )
117

    
118
  if ieio == constants.IEIO_SCRIPT:
119
    assert len(ieioargs) == 2
120
    return (objects.Disk.FromDict(ieioargs[0]), ieioargs[1])
121

    
122
  return ieioargs
123

    
124

    
125
def _DefaultAlternative(value, default):
126
  """Returns value or, if evaluating to False, a default value.
127

128
  Returns the given value, unless it evaluates to False. In the latter case the
129
  default value is returned.
130

131
  @param value: Value to return if it doesn't evaluate to False
132
  @param default: Default value
133
  @return: Given value or the default
134

135
  """
136
  if value:
137
    return value
138

    
139
  return default
140

    
141

    
142
class MlockallRequestExecutor(http.server.HttpServerRequestExecutor):
143
  """Subclass ensuring request handlers are locked in RAM.
144

145
  """
146
  def __init__(self, *args, **kwargs):
147
    utils.Mlockall()
148

    
149
    http.server.HttpServerRequestExecutor.__init__(self, *args, **kwargs)
150

    
151

    
152
class NodeRequestHandler(http.server.HttpServerHandler):
153
  """The server implementation.
154

155
  This class holds all methods exposed over the RPC interface.
156

157
  """
158
  # too many public methods, and unused args - all methods get params
159
  # due to the API
160
  # pylint: disable=R0904,W0613
161
  def __init__(self):
162
    http.server.HttpServerHandler.__init__(self)
163
    self.noded_pid = os.getpid()
164

    
165
  def HandleRequest(self, req):
166
    """Handle a request.
167

168
    """
169
    if req.request_method.upper() != http.HTTP_POST:
170
      raise http.HttpBadRequest("Only the POST method is supported")
171

    
172
    path = req.request_path
173
    if path.startswith("/"):
174
      path = path[1:]
175

    
176
    method = getattr(self, "perspective_%s" % path, None)
177
    if method is None:
178
      raise http.HttpNotFound()
179

    
180
    try:
181
      result = (True, method(serializer.LoadJson(req.request_body)))
182

    
183
    except backend.RPCFail, err:
184
      # our custom failure exception; str(err) works fine if the
185
      # exception was constructed with a single argument, and in
186
      # this case, err.message == err.args[0] == str(err)
187
      result = (False, str(err))
188
    except errors.QuitGanetiException, err:
189
      # Tell parent to quit
190
      logging.info("Shutting down the node daemon, arguments: %s",
191
                   str(err.args))
192
      os.kill(self.noded_pid, signal.SIGTERM)
193
      # And return the error's arguments, which must be already in
194
      # correct tuple format
195
      result = err.args
196
    except Exception, err:
197
      logging.exception("Error in RPC call")
198
      result = (False, "Error while executing backend function: %s" % str(err))
199

    
200
    return serializer.DumpJson(result)
201

    
202
  # the new block devices  --------------------------
203

    
204
  @staticmethod
205
  def perspective_blockdev_create(params):
206
    """Create a block device.
207

208
    """
209
    (bdev_s, size, owner, on_primary, info, excl_stor) = params
210
    bdev = objects.Disk.FromDict(bdev_s)
211
    if bdev is None:
212
      raise ValueError("can't unserialize data!")
213
    return backend.BlockdevCreate(bdev, size, owner, on_primary, info,
214
                                  excl_stor)
215

    
216
  @staticmethod
217
  def perspective_blockdev_pause_resume_sync(params):
218
    """Pause/resume sync of a block device.
219

220
    """
221
    disks_s, pause = params
222
    disks = [objects.Disk.FromDict(bdev_s) for bdev_s in disks_s]
223
    return backend.BlockdevPauseResumeSync(disks, pause)
224

    
225
  @staticmethod
226
  def perspective_blockdev_wipe(params):
227
    """Wipe a block device.
228

229
    """
230
    bdev_s, offset, size = params
231
    bdev = objects.Disk.FromDict(bdev_s)
232
    return backend.BlockdevWipe(bdev, offset, size)
233

    
234
  @staticmethod
235
  def perspective_blockdev_remove(params):
236
    """Remove a block device.
237

238
    """
239
    bdev_s = params[0]
240
    bdev = objects.Disk.FromDict(bdev_s)
241
    return backend.BlockdevRemove(bdev)
242

    
243
  @staticmethod
244
  def perspective_blockdev_rename(params):
245
    """Remove a block device.
246

247
    """
248
    devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params[0]]
249
    return backend.BlockdevRename(devlist)
250

    
251
  @staticmethod
252
  def perspective_blockdev_assemble(params):
253
    """Assemble a block device.
254

255
    """
256
    bdev_s, owner, on_primary, idx = params
257
    bdev = objects.Disk.FromDict(bdev_s)
258
    if bdev is None:
259
      raise ValueError("can't unserialize data!")
260
    return backend.BlockdevAssemble(bdev, owner, on_primary, idx)
261

    
262
  @staticmethod
263
  def perspective_blockdev_shutdown(params):
264
    """Shutdown a block device.
265

266
    """
267
    bdev_s = params[0]
268
    bdev = objects.Disk.FromDict(bdev_s)
269
    if bdev is None:
270
      raise ValueError("can't unserialize data!")
271
    return backend.BlockdevShutdown(bdev)
272

    
273
  @staticmethod
274
  def perspective_blockdev_addchildren(params):
275
    """Add a child to a mirror device.
276

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

280
    """
281
    bdev_s, ndev_s = params
282
    bdev = objects.Disk.FromDict(bdev_s)
283
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
284
    if bdev is None or ndevs.count(None) > 0:
285
      raise ValueError("can't unserialize data!")
286
    return backend.BlockdevAddchildren(bdev, ndevs)
287

    
288
  @staticmethod
289
  def perspective_blockdev_removechildren(params):
290
    """Remove a child from a mirror device.
291

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

295
    """
296
    bdev_s, ndev_s = params
297
    bdev = objects.Disk.FromDict(bdev_s)
298
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
299
    if bdev is None or ndevs.count(None) > 0:
300
      raise ValueError("can't unserialize data!")
301
    return backend.BlockdevRemovechildren(bdev, ndevs)
302

    
303
  @staticmethod
304
  def perspective_blockdev_getmirrorstatus(params):
305
    """Return the mirror status for a list of disks.
306

307
    """
308
    disks = [objects.Disk.FromDict(dsk_s)
309
             for dsk_s in params[0]]
310
    return [status.ToDict()
311
            for status in backend.BlockdevGetmirrorstatus(disks)]
312

    
313
  @staticmethod
314
  def perspective_blockdev_getmirrorstatus_multi(params):
315
    """Return the mirror status for a list of disks.
316

317
    """
318
    (node_disks, ) = params
319

    
320
    disks = [objects.Disk.FromDict(dsk_s) for dsk_s in node_disks]
321

    
322
    result = []
323

    
324
    for (success, status) in backend.BlockdevGetmirrorstatusMulti(disks):
325
      if success:
326
        result.append((success, status.ToDict()))
327
      else:
328
        result.append((success, status))
329

    
330
    return result
331

    
332
  @staticmethod
333
  def perspective_blockdev_find(params):
334
    """Expose the FindBlockDevice functionality for a disk.
335

336
    This will try to find but not activate a disk.
337

338
    """
339
    disk = objects.Disk.FromDict(params[0])
340

    
341
    result = backend.BlockdevFind(disk)
342
    if result is None:
343
      return None
344

    
345
    return result.ToDict()
346

    
347
  @staticmethod
348
  def perspective_blockdev_snapshot(params):
349
    """Create a snapshot device.
350

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

355
    """
356
    cfbd = objects.Disk.FromDict(params[0])
357
    return backend.BlockdevSnapshot(cfbd)
358

    
359
  @staticmethod
360
  def perspective_blockdev_grow(params):
361
    """Grow a stack of devices.
362

363
    """
364
    if len(params) < 5:
365
      raise ValueError("Received only %s parameters in blockdev_grow,"
366
                       " old master?" % len(params))
367
    cfbd = objects.Disk.FromDict(params[0])
368
    amount = params[1]
369
    dryrun = params[2]
370
    backingstore = params[3]
371
    excl_stor = params[4]
372
    return backend.BlockdevGrow(cfbd, amount, dryrun, backingstore, excl_stor)
373

    
374
  @staticmethod
375
  def perspective_blockdev_close(params):
376
    """Closes the given block devices.
377

378
    """
379
    disks = [objects.Disk.FromDict(cf) for cf in params[1]]
380
    return backend.BlockdevClose(params[0], disks)
381

    
382
  @staticmethod
383
  def perspective_blockdev_getdimensions(params):
384
    """Compute the sizes of the given block devices.
385

386
    """
387
    disks = [objects.Disk.FromDict(cf) for cf in params[0]]
388
    return backend.BlockdevGetdimensions(disks)
389

    
390
  @staticmethod
391
  def perspective_blockdev_export(params):
392
    """Compute the sizes of the given block devices.
393

394
    """
395
    disk = objects.Disk.FromDict(params[0])
396
    dest_node, dest_path, cluster_name = params[1:]
397
    return backend.BlockdevExport(disk, dest_node, dest_path, cluster_name)
398

    
399
  @staticmethod
400
  def perspective_blockdev_setinfo(params):
401
    """Sets metadata information on the given block device.
402

403
    """
404
    (disk, info) = params
405
    disk = objects.Disk.FromDict(disk)
406
    return backend.BlockdevSetInfo(disk, info)
407

    
408
  # blockdev/drbd specific methods ----------
409

    
410
  @staticmethod
411
  def perspective_drbd_disconnect_net(params):
412
    """Disconnects the network connection of drbd disks.
413

414
    Note that this is only valid for drbd disks, so the members of the
415
    disk list must all be drbd devices.
416

417
    """
418
    nodes_ip, disks, target_node_uuid = params
419
    disks = [objects.Disk.FromDict(cf) for cf in disks]
420
    return backend.DrbdDisconnectNet(target_node_uuid, nodes_ip, disks)
421

    
422
  @staticmethod
423
  def perspective_drbd_attach_net(params):
424
    """Attaches the network connection of drbd disks.
425

426
    Note that this is only valid for drbd disks, so the members of the
427
    disk list must all be drbd devices.
428

429
    """
430
    nodes_ip, disks, instance_name, multimaster, target_node_uuid = params
431
    disks = [objects.Disk.FromDict(cf) for cf in disks]
432
    return backend.DrbdAttachNet(target_node_uuid, nodes_ip, disks,
433
                                 instance_name, multimaster)
434

    
435
  @staticmethod
436
  def perspective_drbd_wait_sync(params):
437
    """Wait until DRBD disks are synched.
438

439
    Note that this is only valid for drbd disks, so the members of the
440
    disk list must all be drbd devices.
441

442
    """
443
    nodes_ip, disks, target_node_uuid = params
444
    disks = [objects.Disk.FromDict(cf) for cf in disks]
445
    return backend.DrbdWaitSync(target_node_uuid, nodes_ip, disks)
446

    
447
  @staticmethod
448
  def perspective_drbd_needs_activation(params):
449
    """Checks if the drbd devices need activation
450

451
    Note that this is only valid for drbd disks, so the members of the
452
    disk list must all be drbd devices.
453

454
    """
455
    nodes_ip, disks, target_node_uuid = params
456
    disks = [objects.Disk.FromDict(cf) for cf in disks]
457
    return backend.DrbdNeedsActivation(target_node_uuid, nodes_ip, disks)
458

    
459
  @staticmethod
460
  def perspective_drbd_helper(params):
461
    """Query drbd helper.
462

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

    
466
  # export/import  --------------------------
467

    
468
  @staticmethod
469
  def perspective_finalize_export(params):
470
    """Expose the finalize export functionality.
471

472
    """
473
    instance = objects.Instance.FromDict(params[0])
474

    
475
    snap_disks = []
476
    for disk in params[1]:
477
      if isinstance(disk, bool):
478
        snap_disks.append(disk)
479
      else:
480
        snap_disks.append(objects.Disk.FromDict(disk))
481

    
482
    return backend.FinalizeExport(instance, snap_disks)
483

    
484
  @staticmethod
485
  def perspective_export_info(params):
486
    """Query information about an existing export on this node.
487

488
    The given path may not contain an export, in which case we return
489
    None.
490

491
    """
492
    path = params[0]
493
    return backend.ExportInfo(path)
494

    
495
  @staticmethod
496
  def perspective_export_list(params):
497
    """List the available exports on this node.
498

499
    Note that as opposed to export_info, which may query data about an
500
    export in any path, this only queries the standard Ganeti path
501
    (pathutils.EXPORT_DIR).
502

503
    """
504
    return backend.ListExports()
505

    
506
  @staticmethod
507
  def perspective_export_remove(params):
508
    """Remove an export.
509

510
    """
511
    export = params[0]
512
    return backend.RemoveExport(export)
513

    
514
  # block device ---------------------
515
  @staticmethod
516
  def perspective_bdev_sizes(params):
517
    """Query the list of block devices
518

519
    """
520
    devices = params[0]
521
    return backend.GetBlockDevSizes(devices)
522

    
523
  # volume  --------------------------
524

    
525
  @staticmethod
526
  def perspective_lv_list(params):
527
    """Query the list of logical volumes in a given volume group.
528

529
    """
530
    vgname = params[0]
531
    return backend.GetVolumeList(vgname)
532

    
533
  @staticmethod
534
  def perspective_vg_list(params):
535
    """Query the list of volume groups.
536

537
    """
538
    return backend.ListVolumeGroups()
539

    
540
  # Storage --------------------------
541

    
542
  @staticmethod
543
  def perspective_storage_list(params):
544
    """Get list of storage units.
545

546
    """
547
    (su_name, su_args, name, fields) = params
548
    return container.GetStorage(su_name, *su_args).List(name, fields)
549

    
550
  @staticmethod
551
  def perspective_storage_modify(params):
552
    """Modify a storage unit.
553

554
    """
555
    (su_name, su_args, name, changes) = params
556
    return container.GetStorage(su_name, *su_args).Modify(name, changes)
557

    
558
  @staticmethod
559
  def perspective_storage_execute(params):
560
    """Execute an operation on a storage unit.
561

562
    """
563
    (su_name, su_args, name, op) = params
564
    return container.GetStorage(su_name, *su_args).Execute(name, op)
565

    
566
  # bridge  --------------------------
567

    
568
  @staticmethod
569
  def perspective_bridges_exist(params):
570
    """Check if all bridges given exist on this node.
571

572
    """
573
    bridges_list = params[0]
574
    return backend.BridgesExist(bridges_list)
575

    
576
  # instance  --------------------------
577

    
578
  @staticmethod
579
  def perspective_instance_os_add(params):
580
    """Install an OS on a given instance.
581

582
    """
583
    inst_s = params[0]
584
    inst = objects.Instance.FromDict(inst_s)
585
    reinstall = params[1]
586
    debug = params[2]
587
    return backend.InstanceOsAdd(inst, reinstall, debug)
588

    
589
  @staticmethod
590
  def perspective_instance_run_rename(params):
591
    """Runs the OS rename script for an instance.
592

593
    """
594
    inst_s, old_name, debug = params
595
    inst = objects.Instance.FromDict(inst_s)
596
    return backend.RunRenameInstance(inst, old_name, debug)
597

    
598
  @staticmethod
599
  def perspective_instance_shutdown(params):
600
    """Shutdown an instance.
601

602
    """
603
    instance = objects.Instance.FromDict(params[0])
604
    timeout = params[1]
605
    trail = params[2]
606
    _extendReasonTrail(trail, "shutdown")
607
    return backend.InstanceShutdown(instance, timeout, trail)
608

    
609
  @staticmethod
610
  def perspective_instance_start(params):
611
    """Start an instance.
612

613
    """
614
    (instance_name, startup_paused, trail) = params
615
    instance = objects.Instance.FromDict(instance_name)
616
    _extendReasonTrail(trail, "start")
617
    return backend.StartInstance(instance, startup_paused, trail)
618

    
619
  @staticmethod
620
  def perspective_migration_info(params):
621
    """Gather information about an instance to be migrated.
622

623
    """
624
    instance = objects.Instance.FromDict(params[0])
625
    return backend.MigrationInfo(instance)
626

    
627
  @staticmethod
628
  def perspective_accept_instance(params):
629
    """Prepare the node to accept an instance.
630

631
    """
632
    instance, info, target = params
633
    instance = objects.Instance.FromDict(instance)
634
    return backend.AcceptInstance(instance, info, target)
635

    
636
  @staticmethod
637
  def perspective_instance_finalize_migration_dst(params):
638
    """Finalize the instance migration on the destination node.
639

640
    """
641
    instance, info, success = params
642
    instance = objects.Instance.FromDict(instance)
643
    return backend.FinalizeMigrationDst(instance, info, success)
644

    
645
  @staticmethod
646
  def perspective_instance_migrate(params):
647
    """Migrates an instance.
648

649
    """
650
    cluster_name, instance, target, live = params
651
    instance = objects.Instance.FromDict(instance)
652
    return backend.MigrateInstance(cluster_name, instance, target, live)
653

    
654
  @staticmethod
655
  def perspective_instance_finalize_migration_src(params):
656
    """Finalize the instance migration on the source node.
657

658
    """
659
    instance, success, live = params
660
    instance = objects.Instance.FromDict(instance)
661
    return backend.FinalizeMigrationSource(instance, success, live)
662

    
663
  @staticmethod
664
  def perspective_instance_get_migration_status(params):
665
    """Reports migration status.
666

667
    """
668
    instance = objects.Instance.FromDict(params[0])
669
    return backend.GetMigrationStatus(instance).ToDict()
670

    
671
  @staticmethod
672
  def perspective_instance_reboot(params):
673
    """Reboot an instance.
674

675
    """
676
    instance = objects.Instance.FromDict(params[0])
677
    reboot_type = params[1]
678
    shutdown_timeout = params[2]
679
    trail = params[3]
680
    _extendReasonTrail(trail, "reboot")
681
    return backend.InstanceReboot(instance, reboot_type, shutdown_timeout,
682
                                  trail)
683

    
684
  @staticmethod
685
  def perspective_instance_balloon_memory(params):
686
    """Modify instance runtime memory.
687

688
    """
689
    instance_dict, memory = params
690
    instance = objects.Instance.FromDict(instance_dict)
691
    return backend.InstanceBalloonMemory(instance, memory)
692

    
693
  @staticmethod
694
  def perspective_instance_info(params):
695
    """Query instance information.
696

697
    """
698
    (instance_name, hypervisor_name, hvparams) = params
699
    return backend.GetInstanceInfo(instance_name, hypervisor_name, hvparams)
700

    
701
  @staticmethod
702
  def perspective_instance_migratable(params):
703
    """Query whether the specified instance can be migrated.
704

705
    """
706
    instance = objects.Instance.FromDict(params[0])
707
    return backend.GetInstanceMigratable(instance)
708

    
709
  @staticmethod
710
  def perspective_all_instances_info(params):
711
    """Query information about all instances.
712

713
    """
714
    (hypervisor_list, all_hvparams) = params
715
    return backend.GetAllInstancesInfo(hypervisor_list, all_hvparams)
716

    
717
  @staticmethod
718
  def perspective_instance_list(params):
719
    """Query the list of running instances.
720

721
    """
722
    (hypervisor_list, hvparams) = params
723
    return backend.GetInstanceList(hypervisor_list, hvparams)
724

    
725
  # node --------------------------
726

    
727
  @staticmethod
728
  def perspective_node_has_ip_address(params):
729
    """Checks if a node has the given ip address.
730

731
    """
732
    return netutils.IPAddress.Own(params[0])
733

    
734
  @staticmethod
735
  def perspective_node_info(params):
736
    """Query node information.
737

738
    """
739
    (storage_units, hv_specs) = params
740
    return backend.GetNodeInfo(storage_units, hv_specs)
741

    
742
  @staticmethod
743
  def perspective_etc_hosts_modify(params):
744
    """Modify a node entry in /etc/hosts.
745

746
    """
747
    backend.EtcHostsModify(params[0], params[1], params[2])
748

    
749
    return True
750

    
751
  @staticmethod
752
  def perspective_node_verify(params):
753
    """Run a verify sequence on this node.
754

755
    """
756
    (what, cluster_name, hvparams) = params
757
    return backend.VerifyNode(what, cluster_name, hvparams)
758

    
759
  @classmethod
760
  def perspective_node_verify_light(cls, params):
761
    """Run a light verify sequence on this node.
762

763
    """
764
    # So far it's the same as the normal node_verify
765
    return cls.perspective_node_verify(params)
766

    
767
  @staticmethod
768
  def perspective_node_start_master_daemons(params):
769
    """Start the master daemons on this node.
770

771
    """
772
    return backend.StartMasterDaemons(params[0])
773

    
774
  @staticmethod
775
  def perspective_node_activate_master_ip(params):
776
    """Activate the master IP on this node.
777

778
    """
779
    master_params = objects.MasterNetworkParameters.FromDict(params[0])
780
    return backend.ActivateMasterIp(master_params, params[1])
781

    
782
  @staticmethod
783
  def perspective_node_deactivate_master_ip(params):
784
    """Deactivate the master IP on this node.
785

786
    """
787
    master_params = objects.MasterNetworkParameters.FromDict(params[0])
788
    return backend.DeactivateMasterIp(master_params, params[1])
789

    
790
  @staticmethod
791
  def perspective_node_stop_master(params):
792
    """Stops master daemons on this node.
793

794
    """
795
    return backend.StopMasterDaemons()
796

    
797
  @staticmethod
798
  def perspective_node_change_master_netmask(params):
799
    """Change the master IP netmask.
800

801
    """
802
    return backend.ChangeMasterNetmask(params[0], params[1], params[2],
803
                                       params[3])
804

    
805
  @staticmethod
806
  def perspective_node_leave_cluster(params):
807
    """Cleanup after leaving a cluster.
808

809
    """
810
    return backend.LeaveCluster(params[0])
811

    
812
  @staticmethod
813
  def perspective_node_volumes(params):
814
    """Query the list of all logical volume groups.
815

816
    """
817
    return backend.NodeVolumes()
818

    
819
  @staticmethod
820
  def perspective_node_demote_from_mc(params):
821
    """Demote a node from the master candidate role.
822

823
    """
824
    return backend.DemoteFromMC()
825

    
826
  @staticmethod
827
  def perspective_node_powercycle(params):
828
    """Tries to powercycle the node.
829

830
    """
831
    (hypervisor_type, hvparams) = params
832
    return backend.PowercycleNode(hypervisor_type, hvparams)
833

    
834
  @staticmethod
835
  def perspective_node_configure_ovs(params):
836
    """Sets up OpenvSwitch on the node.
837

838
    """
839
    (ovs_name, ovs_link) = params
840
    return backend.ConfigureOVS(ovs_name, ovs_link)
841

    
842
  # cluster --------------------------
843

    
844
  @staticmethod
845
  def perspective_version(params):
846
    """Query version information.
847

848
    """
849
    return constants.PROTOCOL_VERSION
850

    
851
  @staticmethod
852
  def perspective_upload_file(params):
853
    """Upload a file.
854

855
    Note that the backend implementation imposes strict rules on which
856
    files are accepted.
857

858
    """
859
    return backend.UploadFile(*(params[0]))
860

    
861
  @staticmethod
862
  def perspective_master_info(params):
863
    """Query master information.
864

865
    """
866
    return backend.GetMasterInfo()
867

    
868
  @staticmethod
869
  def perspective_run_oob(params):
870
    """Runs oob on node.
871

872
    """
873
    output = backend.RunOob(params[0], params[1], params[2], params[3])
874
    if output:
875
      result = serializer.LoadJson(output)
876
    else:
877
      result = None
878
    return result
879

    
880
  @staticmethod
881
  def perspective_restricted_command(params):
882
    """Runs a restricted command.
883

884
    """
885
    (cmd, ) = params
886

    
887
    return backend.RunRestrictedCmd(cmd)
888

    
889
  @staticmethod
890
  def perspective_write_ssconf_files(params):
891
    """Write ssconf files.
892

893
    """
894
    (values,) = params
895
    return ssconf.WriteSsconfFiles(values)
896

    
897
  @staticmethod
898
  def perspective_get_watcher_pause(params):
899
    """Get watcher pause end.
900

901
    """
902
    return utils.ReadWatcherPauseFile(pathutils.WATCHER_PAUSEFILE)
903

    
904
  @staticmethod
905
  def perspective_set_watcher_pause(params):
906
    """Set watcher pause.
907

908
    """
909
    (until, ) = params
910
    return backend.SetWatcherPause(until)
911

    
912
  # os -----------------------
913

    
914
  @staticmethod
915
  def perspective_os_diagnose(params):
916
    """Query detailed information about existing OSes.
917

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

    
921
  @staticmethod
922
  def perspective_os_get(params):
923
    """Query information about a given OS.
924

925
    """
926
    name = params[0]
927
    os_obj = backend.OSFromDisk(name)
928
    return os_obj.ToDict()
929

    
930
  @staticmethod
931
  def perspective_os_validate(params):
932
    """Run a given OS' validation routine.
933

934
    """
935
    required, name, checks, params = params
936
    return backend.ValidateOS(required, name, checks, params)
937

    
938
  # extstorage -----------------------
939

    
940
  @staticmethod
941
  def perspective_extstorage_diagnose(params):
942
    """Query detailed information about existing extstorage providers.
943

944
    """
945
    return backend.DiagnoseExtStorage()
946

    
947
  # hooks -----------------------
948

    
949
  @staticmethod
950
  def perspective_hooks_runner(params):
951
    """Run hook scripts.
952

953
    """
954
    hpath, phase, env = params
955
    hr = backend.HooksRunner()
956
    return hr.RunHooks(hpath, phase, env)
957

    
958
  # iallocator -----------------
959

    
960
  @staticmethod
961
  def perspective_iallocator_runner(params):
962
    """Run an iallocator script.
963

964
    """
965
    name, idata = params
966
    iar = backend.IAllocatorRunner()
967
    return iar.Run(name, idata)
968

    
969
  # test -----------------------
970

    
971
  @staticmethod
972
  def perspective_test_delay(params):
973
    """Run test delay.
974

975
    """
976
    duration = params[0]
977
    status, rval = utils.TestDelay(duration)
978
    if not status:
979
      raise backend.RPCFail(rval)
980
    return rval
981

    
982
  # file storage ---------------
983

    
984
  @staticmethod
985
  def perspective_file_storage_dir_create(params):
986
    """Create the file storage directory.
987

988
    """
989
    file_storage_dir = params[0]
990
    return backend.CreateFileStorageDir(file_storage_dir)
991

    
992
  @staticmethod
993
  def perspective_file_storage_dir_remove(params):
994
    """Remove the file storage directory.
995

996
    """
997
    file_storage_dir = params[0]
998
    return backend.RemoveFileStorageDir(file_storage_dir)
999

    
1000
  @staticmethod
1001
  def perspective_file_storage_dir_rename(params):
1002
    """Rename the file storage directory.
1003

1004
    """
1005
    old_file_storage_dir = params[0]
1006
    new_file_storage_dir = params[1]
1007
    return backend.RenameFileStorageDir(old_file_storage_dir,
1008
                                        new_file_storage_dir)
1009

    
1010
  # jobs ------------------------
1011

    
1012
  @staticmethod
1013
  @_RequireJobQueueLock
1014
  def perspective_jobqueue_update(params):
1015
    """Update job queue.
1016

1017
    """
1018
    (file_name, content) = params
1019
    return backend.JobQueueUpdate(file_name, content)
1020

    
1021
  @staticmethod
1022
  @_RequireJobQueueLock
1023
  def perspective_jobqueue_purge(params):
1024
    """Purge job queue.
1025

1026
    """
1027
    return backend.JobQueuePurge()
1028

    
1029
  @staticmethod
1030
  @_RequireJobQueueLock
1031
  def perspective_jobqueue_rename(params):
1032
    """Rename a job queue file.
1033

1034
    """
1035
    # TODO: What if a file fails to rename?
1036
    return [backend.JobQueueRename(old, new) for old, new in params[0]]
1037

    
1038
  @staticmethod
1039
  @_RequireJobQueueLock
1040
  def perspective_jobqueue_set_drain_flag(params):
1041
    """Set job queue's drain flag.
1042

1043
    """
1044
    (flag, ) = params
1045

    
1046
    return jstore.SetDrainFlag(flag)
1047

    
1048
  # hypervisor ---------------
1049

    
1050
  @staticmethod
1051
  def perspective_hypervisor_validate_params(params):
1052
    """Validate the hypervisor parameters.
1053

1054
    """
1055
    (hvname, hvparams) = params
1056
    return backend.ValidateHVParams(hvname, hvparams)
1057

    
1058
  # Crypto
1059

    
1060
  @staticmethod
1061
  def perspective_x509_cert_create(params):
1062
    """Creates a new X509 certificate for SSL/TLS.
1063

1064
    """
1065
    (validity, ) = params
1066
    return backend.CreateX509Certificate(validity)
1067

    
1068
  @staticmethod
1069
  def perspective_x509_cert_remove(params):
1070
    """Removes a X509 certificate.
1071

1072
    """
1073
    (name, ) = params
1074
    return backend.RemoveX509Certificate(name)
1075

    
1076
  # Import and export
1077

    
1078
  @staticmethod
1079
  def perspective_import_start(params):
1080
    """Starts an import daemon.
1081

1082
    """
1083
    (opts_s, instance, component, (dest, dest_args)) = params
1084

    
1085
    opts = objects.ImportExportOptions.FromDict(opts_s)
1086

    
1087
    return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
1088
                                           None, None,
1089
                                           objects.Instance.FromDict(instance),
1090
                                           component, dest,
1091
                                           _DecodeImportExportIO(dest,
1092
                                                                 dest_args))
1093

    
1094
  @staticmethod
1095
  def perspective_export_start(params):
1096
    """Starts an export daemon.
1097

1098
    """
1099
    (opts_s, host, port, instance, component, (source, source_args)) = params
1100

    
1101
    opts = objects.ImportExportOptions.FromDict(opts_s)
1102

    
1103
    return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
1104
                                           host, port,
1105
                                           objects.Instance.FromDict(instance),
1106
                                           component, source,
1107
                                           _DecodeImportExportIO(source,
1108
                                                                 source_args))
1109

    
1110
  @staticmethod
1111
  def perspective_impexp_status(params):
1112
    """Retrieves the status of an import or export daemon.
1113

1114
    """
1115
    return backend.GetImportExportStatus(params[0])
1116

    
1117
  @staticmethod
1118
  def perspective_impexp_abort(params):
1119
    """Aborts an import or export.
1120

1121
    """
1122
    return backend.AbortImportExport(params[0])
1123

    
1124
  @staticmethod
1125
  def perspective_impexp_cleanup(params):
1126
    """Cleans up after an import or export.
1127

1128
    """
1129
    return backend.CleanupImportExport(params[0])
1130

    
1131

    
1132
def CheckNoded(_, args):
1133
  """Initial checks whether to run or exit with a failure.
1134

1135
  """
1136
  if args: # noded doesn't take any arguments
1137
    print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
1138
                          sys.argv[0])
1139
    sys.exit(constants.EXIT_FAILURE)
1140
  try:
1141
    codecs.lookup("string-escape")
1142
  except LookupError:
1143
    print >> sys.stderr, ("Can't load the string-escape code which is part"
1144
                          " of the Python installation. Is your installation"
1145
                          " complete/correct? Aborting.")
1146
    sys.exit(constants.EXIT_FAILURE)
1147

    
1148

    
1149
def PrepNoded(options, _):
1150
  """Preparation node daemon function, executed with the PID file held.
1151

1152
  """
1153
  if options.mlock:
1154
    request_executor_class = MlockallRequestExecutor
1155
    try:
1156
      utils.Mlockall()
1157
    except errors.NoCtypesError:
1158
      logging.warning("Cannot set memory lock, ctypes module not found")
1159
      request_executor_class = http.server.HttpServerRequestExecutor
1160
  else:
1161
    request_executor_class = http.server.HttpServerRequestExecutor
1162

    
1163
  # Read SSL certificate
1164
  if options.ssl:
1165
    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
1166
                                    ssl_cert_path=options.ssl_cert)
1167
  else:
1168
    ssl_params = None
1169

    
1170
  err = _PrepareQueueLock()
1171
  if err is not None:
1172
    # this might be some kind of file-system/permission error; while
1173
    # this breaks the job queue functionality, we shouldn't prevent
1174
    # startup of the whole node daemon because of this
1175
    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
1176

    
1177
  handler = NodeRequestHandler()
1178

    
1179
  mainloop = daemon.Mainloop()
1180
  server = \
1181
    http.server.HttpServer(mainloop, options.bind_address, options.port,
1182
                           handler, ssl_params=ssl_params, ssl_verify_peer=True,
1183
                           request_executor_class=request_executor_class)
1184
  server.Start()
1185

    
1186
  return (mainloop, server)
1187

    
1188

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

1192
  """
1193
  (mainloop, server) = prep_data
1194
  try:
1195
    mainloop.Run()
1196
  finally:
1197
    server.Stop()
1198

    
1199

    
1200
def Main():
1201
  """Main function for the node daemon.
1202

1203
  """
1204
  parser = OptionParser(description="Ganeti node daemon",
1205
                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]\
1206
                              \ [-i INTERFACE]",
1207
                        version="%%prog (ganeti) %s" %
1208
                        constants.RELEASE_VERSION)
1209
  parser.add_option("--no-mlock", dest="mlock",
1210
                    help="Do not mlock the node memory in ram",
1211
                    default=True, action="store_false")
1212

    
1213
  daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded,
1214
                     default_ssl_cert=pathutils.NODED_CERT_FILE,
1215
                     default_ssl_key=pathutils.NODED_CERT_FILE,
1216
                     console_logging=True)