Statistics
| Branch: | Tag: | Revision:

root / lib / server / noded.py @ b8c160c1

History | View | Annotate | Download (28.7 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2010, 2011 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

    
52
import ganeti.http.server # pylint: disable=W0611
53

    
54

    
55
queue_lock = None
56

    
57

    
58
def _PrepareQueueLock():
59
  """Try to prepare the queue lock.
60

61
  @return: None for success, otherwise an exception object
62

63
  """
64
  global queue_lock # pylint: disable=W0603
65

    
66
  if queue_lock is not None:
67
    return None
68

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

    
76

    
77
def _RequireJobQueueLock(fn):
78
  """Decorator for job queue manipulating functions.
79

80
  """
81
  QUEUE_LOCK_TIMEOUT = 10
82

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

    
95
  return wrapper
96

    
97

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

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

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

    
110
  return ieioargs
111

    
112

    
113
class MlockallRequestExecutor(http.server.HttpServerRequestExecutor):
114
  """Custom Request Executor class that ensures NodeHttpServer children are
115
  locked in ram.
116

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

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

    
123

    
124
class NodeHttpServer(http.server.HttpServer):
125
  """The server implementation.
126

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

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

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

140
    """
141
    if req.request_method.upper() != http.HTTP_PUT:
142
      raise http.HttpBadRequest()
143

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

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

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

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

    
172
    return serializer.DumpJson(result, indent=False)
173

    
174
  # the new block devices  --------------------------
175

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

288
    """
289
    (node_disks, ) = params
290

    
291
    node_name = netutils.Hostname.GetSysName()
292

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

    
296
    result = []
297

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

    
304
    return result
305

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

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

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

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

    
319
    return result.ToDict()
320

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

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

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

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

337
    """
338
    cfbd = objects.Disk.FromDict(params[0])
339
    amount = params[1]
340
    dryrun = params[2]
341
    return backend.BlockdevGrow(cfbd, amount, dryrun)
342

    
343
  @staticmethod
344
  def perspective_blockdev_close(params):
345
    """Closes the given block devices.
346

347
    """
348
    disks = [objects.Disk.FromDict(cf) for cf in params[1]]
349
    return backend.BlockdevClose(params[0], disks)
350

    
351
  @staticmethod
352
  def perspective_blockdev_getsize(params):
353
    """Compute the sizes of the given block devices.
354

355
    """
356
    disks = [objects.Disk.FromDict(cf) for cf in params[0]]
357
    return backend.BlockdevGetsize(disks)
358

    
359
  @staticmethod
360
  def perspective_blockdev_export(params):
361
    """Compute the sizes of the given block devices.
362

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

    
368
  # blockdev/drbd specific methods ----------
369

    
370
  @staticmethod
371
  def perspective_drbd_disconnect_net(params):
372
    """Disconnects the network connection of drbd disks.
373

374
    Note that this is only valid for drbd disks, so the members of the
375
    disk list must all be drbd devices.
376

377
    """
378
    nodes_ip, disks = params
379
    disks = [objects.Disk.FromDict(cf) for cf in disks]
380
    return backend.DrbdDisconnectNet(nodes_ip, disks)
381

    
382
  @staticmethod
383
  def perspective_drbd_attach_net(params):
384
    """Attaches 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, instance_name, multimaster = params
391
    disks = [objects.Disk.FromDict(cf) for cf in disks]
392
    return backend.DrbdAttachNet(nodes_ip, disks,
393
                                     instance_name, multimaster)
394

    
395
  @staticmethod
396
  def perspective_drbd_wait_sync(params):
397
    """Wait until DRBD disks are synched.
398

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

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

    
407
  @staticmethod
408
  def perspective_drbd_helper(params):
409
    """Query drbd helper.
410

411
    """
412
    return backend.GetDrbdUsermodeHelper()
413

    
414
  # export/import  --------------------------
415

    
416
  @staticmethod
417
  def perspective_finalize_export(params):
418
    """Expose the finalize export functionality.
419

420
    """
421
    instance = objects.Instance.FromDict(params[0])
422

    
423
    snap_disks = []
424
    for disk in params[1]:
425
      if isinstance(disk, bool):
426
        snap_disks.append(disk)
427
      else:
428
        snap_disks.append(objects.Disk.FromDict(disk))
429

    
430
    return backend.FinalizeExport(instance, snap_disks)
431

    
432
  @staticmethod
433
  def perspective_export_info(params):
434
    """Query information about an existing export on this node.
435

436
    The given path may not contain an export, in which case we return
437
    None.
438

439
    """
440
    path = params[0]
441
    return backend.ExportInfo(path)
442

    
443
  @staticmethod
444
  def perspective_export_list(params):
445
    """List the available exports on this node.
446

447
    Note that as opposed to export_info, which may query data about an
448
    export in any path, this only queries the standard Ganeti path
449
    (constants.EXPORT_DIR).
450

451
    """
452
    return backend.ListExports()
453

    
454
  @staticmethod
455
  def perspective_export_remove(params):
456
    """Remove an export.
457

458
    """
459
    export = params[0]
460
    return backend.RemoveExport(export)
461

    
462
  # block device ---------------------
463
  @staticmethod
464
  def perspective_bdev_sizes(params):
465
    """Query the list of block devices
466

467
    """
468
    devices = params[0]
469
    return backend.GetBlockDevSizes(devices)
470

    
471
  # volume  --------------------------
472

    
473
  @staticmethod
474
  def perspective_lv_list(params):
475
    """Query the list of logical volumes in a given volume group.
476

477
    """
478
    vgname = params[0]
479
    return backend.GetVolumeList(vgname)
480

    
481
  @staticmethod
482
  def perspective_vg_list(params):
483
    """Query the list of volume groups.
484

485
    """
486
    return backend.ListVolumeGroups()
487

    
488
  # Storage --------------------------
489

    
490
  @staticmethod
491
  def perspective_storage_list(params):
492
    """Get list of storage units.
493

494
    """
495
    (su_name, su_args, name, fields) = params
496
    return storage.GetStorage(su_name, *su_args).List(name, fields)
497

    
498
  @staticmethod
499
  def perspective_storage_modify(params):
500
    """Modify a storage unit.
501

502
    """
503
    (su_name, su_args, name, changes) = params
504
    return storage.GetStorage(su_name, *su_args).Modify(name, changes)
505

    
506
  @staticmethod
507
  def perspective_storage_execute(params):
508
    """Execute an operation on a storage unit.
509

510
    """
511
    (su_name, su_args, name, op) = params
512
    return storage.GetStorage(su_name, *su_args).Execute(name, op)
513

    
514
  # bridge  --------------------------
515

    
516
  @staticmethod
517
  def perspective_bridges_exist(params):
518
    """Check if all bridges given exist on this node.
519

520
    """
521
    bridges_list = params[0]
522
    return backend.BridgesExist(bridges_list)
523

    
524
  # instance  --------------------------
525

    
526
  @staticmethod
527
  def perspective_instance_os_add(params):
528
    """Install an OS on a given instance.
529

530
    """
531
    inst_s = params[0]
532
    inst = objects.Instance.FromDict(inst_s)
533
    reinstall = params[1]
534
    debug = params[2]
535
    return backend.InstanceOsAdd(inst, reinstall, debug)
536

    
537
  @staticmethod
538
  def perspective_instance_run_rename(params):
539
    """Runs the OS rename script for an instance.
540

541
    """
542
    inst_s, old_name, debug = params
543
    inst = objects.Instance.FromDict(inst_s)
544
    return backend.RunRenameInstance(inst, old_name, debug)
545

    
546
  @staticmethod
547
  def perspective_instance_shutdown(params):
548
    """Shutdown an instance.
549

550
    """
551
    instance = objects.Instance.FromDict(params[0])
552
    timeout = params[1]
553
    return backend.InstanceShutdown(instance, timeout)
554

    
555
  @staticmethod
556
  def perspective_instance_start(params):
557
    """Start an instance.
558

559
    """
560
    (instance_name, startup_paused) = params
561
    instance = objects.Instance.FromDict(instance_name)
562
    return backend.StartInstance(instance, startup_paused)
563

    
564
  @staticmethod
565
  def perspective_migration_info(params):
566
    """Gather information about an instance to be migrated.
567

568
    """
569
    instance = objects.Instance.FromDict(params[0])
570
    return backend.MigrationInfo(instance)
571

    
572
  @staticmethod
573
  def perspective_accept_instance(params):
574
    """Prepare the node to accept an instance.
575

576
    """
577
    instance, info, target = params
578
    instance = objects.Instance.FromDict(instance)
579
    return backend.AcceptInstance(instance, info, target)
580

    
581
  @staticmethod
582
  def perspective_instance_finalize_migration_dst(params):
583
    """Finalize the instance migration on the destination node.
584

585
    """
586
    instance, info, success = params
587
    instance = objects.Instance.FromDict(instance)
588
    return backend.FinalizeMigrationDst(instance, info, success)
589

    
590
  @staticmethod
591
  def perspective_instance_migrate(params):
592
    """Migrates an instance.
593

594
    """
595
    instance, target, live = params
596
    instance = objects.Instance.FromDict(instance)
597
    return backend.MigrateInstance(instance, target, live)
598

    
599
  @staticmethod
600
  def perspective_instance_finalize_migration_src(params):
601
    """Finalize the instance migration on the source node.
602

603
    """
604
    instance, success, live = params
605
    instance = objects.Instance.FromDict(instance)
606
    return backend.FinalizeMigrationSource(instance, success, live)
607

    
608
  @staticmethod
609
  def perspective_instance_get_migration_status(params):
610
    """Reports migration status.
611

612
    """
613
    instance = objects.Instance.FromDict(params[0])
614
    return backend.GetMigrationStatus(instance).ToDict()
615

    
616
  @staticmethod
617
  def perspective_instance_reboot(params):
618
    """Reboot an instance.
619

620
    """
621
    instance = objects.Instance.FromDict(params[0])
622
    reboot_type = params[1]
623
    shutdown_timeout = params[2]
624
    return backend.InstanceReboot(instance, reboot_type, shutdown_timeout)
625

    
626
  @staticmethod
627
  def perspective_instance_info(params):
628
    """Query instance information.
629

630
    """
631
    return backend.GetInstanceInfo(params[0], params[1])
632

    
633
  @staticmethod
634
  def perspective_instance_migratable(params):
635
    """Query whether the specified instance can be migrated.
636

637
    """
638
    instance = objects.Instance.FromDict(params[0])
639
    return backend.GetInstanceMigratable(instance)
640

    
641
  @staticmethod
642
  def perspective_all_instances_info(params):
643
    """Query information about all instances.
644

645
    """
646
    return backend.GetAllInstancesInfo(params[0])
647

    
648
  @staticmethod
649
  def perspective_instance_list(params):
650
    """Query the list of running instances.
651

652
    """
653
    return backend.GetInstanceList(params[0])
654

    
655
  # node --------------------------
656

    
657
  @staticmethod
658
  def perspective_node_has_ip_address(params):
659
    """Checks if a node has the given ip address.
660

661
    """
662
    return netutils.IPAddress.Own(params[0])
663

    
664
  @staticmethod
665
  def perspective_node_info(params):
666
    """Query node information.
667

668
    """
669
    vgname, hypervisor_type = params
670
    return backend.GetNodeInfo(vgname, hypervisor_type)
671

    
672
  @staticmethod
673
  def perspective_etc_hosts_modify(params):
674
    """Modify a node entry in /etc/hosts.
675

676
    """
677
    backend.EtcHostsModify(params[0], params[1], params[2])
678

    
679
    return True
680

    
681
  @staticmethod
682
  def perspective_node_verify(params):
683
    """Run a verify sequence on this node.
684

685
    """
686
    return backend.VerifyNode(params[0], params[1])
687

    
688
  @staticmethod
689
  def perspective_node_start_master_daemons(params):
690
    """Start the master daemons on this node.
691

692
    """
693
    return backend.StartMasterDaemons(params[0])
694

    
695
  @staticmethod
696
  def perspective_node_activate_master_ip(params):
697
    """Activate the master IP on this node.
698

699
    """
700
    return backend.ActivateMasterIp()
701

    
702
  @staticmethod
703
  def perspective_node_deactivate_master_ip(params):
704
    """Deactivate the master IP on this node.
705

706
    """
707
    return backend.DeactivateMasterIp()
708

    
709
  @staticmethod
710
  def perspective_node_stop_master(params):
711
    """Deactivate the master IP and stops master daemons on this node.
712

713
    Sometimes both operations need to be executed at the same time (doing one of
714
    the two would make impossible to do the other one).
715

716
    """
717
    backend.DeactivateMasterIp()
718
    return backend.StopMasterDaemons()
719

    
720
  @staticmethod
721
  def perspective_node_change_master_netmask(params):
722
    """Change the master IP netmask.
723

724
    """
725
    return backend.ChangeMasterNetmask(params[0])
726

    
727
  @staticmethod
728
  def perspective_node_leave_cluster(params):
729
    """Cleanup after leaving a cluster.
730

731
    """
732
    return backend.LeaveCluster(params[0])
733

    
734
  @staticmethod
735
  def perspective_node_volumes(params):
736
    """Query the list of all logical volume groups.
737

738
    """
739
    return backend.NodeVolumes()
740

    
741
  @staticmethod
742
  def perspective_node_demote_from_mc(params):
743
    """Demote a node from the master candidate role.
744

745
    """
746
    return backend.DemoteFromMC()
747

    
748
  @staticmethod
749
  def perspective_node_powercycle(params):
750
    """Tries to powercycle the nod.
751

752
    """
753
    hypervisor_type = params[0]
754
    return backend.PowercycleNode(hypervisor_type)
755

    
756
  # cluster --------------------------
757

    
758
  @staticmethod
759
  def perspective_version(params):
760
    """Query version information.
761

762
    """
763
    return constants.PROTOCOL_VERSION
764

    
765
  @staticmethod
766
  def perspective_upload_file(params):
767
    """Upload a file.
768

769
    Note that the backend implementation imposes strict rules on which
770
    files are accepted.
771

772
    """
773
    return backend.UploadFile(*(params[0]))
774

    
775
  @staticmethod
776
  def perspective_master_info(params):
777
    """Query master information.
778

779
    """
780
    return backend.GetMasterInfo()
781

    
782
  @staticmethod
783
  def perspective_run_oob(params):
784
    """Runs oob on node.
785

786
    """
787
    output = backend.RunOob(params[0], params[1], params[2], params[3])
788
    if output:
789
      result = serializer.LoadJson(output)
790
    else:
791
      result = None
792
    return result
793

    
794
  @staticmethod
795
  def perspective_write_ssconf_files(params):
796
    """Write ssconf files.
797

798
    """
799
    (values,) = params
800
    return backend.WriteSsconfFiles(values)
801

    
802
  # os -----------------------
803

    
804
  @staticmethod
805
  def perspective_os_diagnose(params):
806
    """Query detailed information about existing OSes.
807

808
    """
809
    return backend.DiagnoseOS()
810

    
811
  @staticmethod
812
  def perspective_os_get(params):
813
    """Query information about a given OS.
814

815
    """
816
    name = params[0]
817
    os_obj = backend.OSFromDisk(name)
818
    return os_obj.ToDict()
819

    
820
  @staticmethod
821
  def perspective_os_validate(params):
822
    """Run a given OS' validation routine.
823

824
    """
825
    required, name, checks, params = params
826
    return backend.ValidateOS(required, name, checks, params)
827

    
828
  # hooks -----------------------
829

    
830
  @staticmethod
831
  def perspective_hooks_runner(params):
832
    """Run hook scripts.
833

834
    """
835
    hpath, phase, env = params
836
    hr = backend.HooksRunner()
837
    return hr.RunHooks(hpath, phase, env)
838

    
839
  # iallocator -----------------
840

    
841
  @staticmethod
842
  def perspective_iallocator_runner(params):
843
    """Run an iallocator script.
844

845
    """
846
    name, idata = params
847
    iar = backend.IAllocatorRunner()
848
    return iar.Run(name, idata)
849

    
850
  # test -----------------------
851

    
852
  @staticmethod
853
  def perspective_test_delay(params):
854
    """Run test delay.
855

856
    """
857
    duration = params[0]
858
    status, rval = utils.TestDelay(duration)
859
    if not status:
860
      raise backend.RPCFail(rval)
861
    return rval
862

    
863
  # file storage ---------------
864

    
865
  @staticmethod
866
  def perspective_file_storage_dir_create(params):
867
    """Create the file storage directory.
868

869
    """
870
    file_storage_dir = params[0]
871
    return backend.CreateFileStorageDir(file_storage_dir)
872

    
873
  @staticmethod
874
  def perspective_file_storage_dir_remove(params):
875
    """Remove the file storage directory.
876

877
    """
878
    file_storage_dir = params[0]
879
    return backend.RemoveFileStorageDir(file_storage_dir)
880

    
881
  @staticmethod
882
  def perspective_file_storage_dir_rename(params):
883
    """Rename the file storage directory.
884

885
    """
886
    old_file_storage_dir = params[0]
887
    new_file_storage_dir = params[1]
888
    return backend.RenameFileStorageDir(old_file_storage_dir,
889
                                        new_file_storage_dir)
890

    
891
  # jobs ------------------------
892

    
893
  @staticmethod
894
  @_RequireJobQueueLock
895
  def perspective_jobqueue_update(params):
896
    """Update job queue.
897

898
    """
899
    (file_name, content) = params
900
    return backend.JobQueueUpdate(file_name, content)
901

    
902
  @staticmethod
903
  @_RequireJobQueueLock
904
  def perspective_jobqueue_purge(params):
905
    """Purge job queue.
906

907
    """
908
    return backend.JobQueuePurge()
909

    
910
  @staticmethod
911
  @_RequireJobQueueLock
912
  def perspective_jobqueue_rename(params):
913
    """Rename a job queue file.
914

915
    """
916
    # TODO: What if a file fails to rename?
917
    return [backend.JobQueueRename(old, new) for old, new in params[0]]
918

    
919
  # hypervisor ---------------
920

    
921
  @staticmethod
922
  def perspective_hypervisor_validate_params(params):
923
    """Validate the hypervisor parameters.
924

925
    """
926
    (hvname, hvparams) = params
927
    return backend.ValidateHVParams(hvname, hvparams)
928

    
929
  # Crypto
930

    
931
  @staticmethod
932
  def perspective_x509_cert_create(params):
933
    """Creates a new X509 certificate for SSL/TLS.
934

935
    """
936
    (validity, ) = params
937
    return backend.CreateX509Certificate(validity)
938

    
939
  @staticmethod
940
  def perspective_x509_cert_remove(params):
941
    """Removes a X509 certificate.
942

943
    """
944
    (name, ) = params
945
    return backend.RemoveX509Certificate(name)
946

    
947
  # Import and export
948

    
949
  @staticmethod
950
  def perspective_import_start(params):
951
    """Starts an import daemon.
952

953
    """
954
    (opts_s, instance, component, (dest, dest_args)) = params
955

    
956
    opts = objects.ImportExportOptions.FromDict(opts_s)
957

    
958
    return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
959
                                           None, None,
960
                                           objects.Instance.FromDict(instance),
961
                                           component, dest,
962
                                           _DecodeImportExportIO(dest,
963
                                                                 dest_args))
964

    
965
  @staticmethod
966
  def perspective_export_start(params):
967
    """Starts an export daemon.
968

969
    """
970
    (opts_s, host, port, instance, component, (source, source_args)) = params
971

    
972
    opts = objects.ImportExportOptions.FromDict(opts_s)
973

    
974
    return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
975
                                           host, port,
976
                                           objects.Instance.FromDict(instance),
977
                                           component, source,
978
                                           _DecodeImportExportIO(source,
979
                                                                 source_args))
980

    
981
  @staticmethod
982
  def perspective_impexp_status(params):
983
    """Retrieves the status of an import or export daemon.
984

985
    """
986
    return backend.GetImportExportStatus(params[0])
987

    
988
  @staticmethod
989
  def perspective_impexp_abort(params):
990
    """Aborts an import or export.
991

992
    """
993
    return backend.AbortImportExport(params[0])
994

    
995
  @staticmethod
996
  def perspective_impexp_cleanup(params):
997
    """Cleans up after an import or export.
998

999
    """
1000
    return backend.CleanupImportExport(params[0])
1001

    
1002

    
1003
def CheckNoded(_, args):
1004
  """Initial checks whether to run or exit with a failure.
1005

1006
  """
1007
  if args: # noded doesn't take any arguments
1008
    print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
1009
                          sys.argv[0])
1010
    sys.exit(constants.EXIT_FAILURE)
1011
  try:
1012
    codecs.lookup("string-escape")
1013
  except LookupError:
1014
    print >> sys.stderr, ("Can't load the string-escape code which is part"
1015
                          " of the Python installation. Is your installation"
1016
                          " complete/correct? Aborting.")
1017
    sys.exit(constants.EXIT_FAILURE)
1018

    
1019

    
1020
def PrepNoded(options, _):
1021
  """Preparation node daemon function, executed with the PID file held.
1022

1023
  """
1024
  if options.mlock:
1025
    request_executor_class = MlockallRequestExecutor
1026
    try:
1027
      utils.Mlockall()
1028
    except errors.NoCtypesError:
1029
      logging.warning("Cannot set memory lock, ctypes module not found")
1030
      request_executor_class = http.server.HttpServerRequestExecutor
1031
  else:
1032
    request_executor_class = http.server.HttpServerRequestExecutor
1033

    
1034
  # Read SSL certificate
1035
  if options.ssl:
1036
    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
1037
                                    ssl_cert_path=options.ssl_cert)
1038
  else:
1039
    ssl_params = None
1040

    
1041
  err = _PrepareQueueLock()
1042
  if err is not None:
1043
    # this might be some kind of file-system/permission error; while
1044
    # this breaks the job queue functionality, we shouldn't prevent
1045
    # startup of the whole node daemon because of this
1046
    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
1047

    
1048
  mainloop = daemon.Mainloop()
1049
  server = NodeHttpServer(mainloop, options.bind_address, options.port,
1050
                          ssl_params=ssl_params, ssl_verify_peer=True,
1051
                          request_executor_class=request_executor_class)
1052
  server.Start()
1053
  return (mainloop, server)
1054

    
1055

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

1059
  """
1060
  (mainloop, server) = prep_data
1061
  try:
1062
    mainloop.Run()
1063
  finally:
1064
    server.Stop()
1065

    
1066

    
1067
def Main():
1068
  """Main function for the node daemon.
1069

1070
  """
1071
  parser = OptionParser(description="Ganeti node daemon",
1072
                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]",
1073
                        version="%%prog (ganeti) %s" %
1074
                        constants.RELEASE_VERSION)
1075
  parser.add_option("--no-mlock", dest="mlock",
1076
                    help="Do not mlock the node memory in ram",
1077
                    default=True, action="store_false")
1078

    
1079
  daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded,
1080
                     default_ssl_cert=constants.NODED_CERT_FILE,
1081
                     default_ssl_key=constants.NODED_CERT_FILE,
1082
                     console_logging=True)