Statistics
| Branch: | Tag: | Revision:

root / lib / server / noded.py @ 91ae95fd

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(params[0], params[1], params[2], params[3])
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(params[0], params[1], params[2],
708
                                      params[3])
709

    
710
  @staticmethod
711
  def perspective_node_stop_master(params):
712
    """Stops master daemons on this node.
713

714
    """
715
    return backend.StopMasterDaemons()
716

    
717
  @staticmethod
718
  def perspective_node_change_master_netmask(params):
719
    """Change the master IP netmask.
720

721
    """
722
    return backend.ChangeMasterNetmask(params[0], params[1], params[2],
723
                                       params[3])
724

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

729
    """
730
    return backend.LeaveCluster(params[0])
731

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

736
    """
737
    return backend.NodeVolumes()
738

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

743
    """
744
    return backend.DemoteFromMC()
745

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

750
    """
751
    hypervisor_type = params[0]
752
    return backend.PowercycleNode(hypervisor_type)
753

    
754
  # cluster --------------------------
755

    
756
  @staticmethod
757
  def perspective_version(params):
758
    """Query version information.
759

760
    """
761
    return constants.PROTOCOL_VERSION
762

    
763
  @staticmethod
764
  def perspective_upload_file(params):
765
    """Upload a file.
766

767
    Note that the backend implementation imposes strict rules on which
768
    files are accepted.
769

770
    """
771
    return backend.UploadFile(*(params[0]))
772

    
773
  @staticmethod
774
  def perspective_master_info(params):
775
    """Query master information.
776

777
    """
778
    return backend.GetMasterInfo()
779

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

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

    
792
  @staticmethod
793
  def perspective_write_ssconf_files(params):
794
    """Write ssconf files.
795

796
    """
797
    (values,) = params
798
    return backend.WriteSsconfFiles(values)
799

    
800
  # os -----------------------
801

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

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

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

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

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

822
    """
823
    required, name, checks, params = params
824
    return backend.ValidateOS(required, name, checks, params)
825

    
826
  # hooks -----------------------
827

    
828
  @staticmethod
829
  def perspective_hooks_runner(params):
830
    """Run hook scripts.
831

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

    
837
  # iallocator -----------------
838

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

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

    
848
  # test -----------------------
849

    
850
  @staticmethod
851
  def perspective_test_delay(params):
852
    """Run test delay.
853

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

    
861
  # file storage ---------------
862

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

867
    """
868
    file_storage_dir = params[0]
869
    return backend.CreateFileStorageDir(file_storage_dir)
870

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

875
    """
876
    file_storage_dir = params[0]
877
    return backend.RemoveFileStorageDir(file_storage_dir)
878

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

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

    
889
  # jobs ------------------------
890

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

896
    """
897
    (file_name, content) = params
898
    return backend.JobQueueUpdate(file_name, content)
899

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

905
    """
906
    return backend.JobQueuePurge()
907

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

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

    
917
  # hypervisor ---------------
918

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

923
    """
924
    (hvname, hvparams) = params
925
    return backend.ValidateHVParams(hvname, hvparams)
926

    
927
  # Crypto
928

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

933
    """
934
    (validity, ) = params
935
    return backend.CreateX509Certificate(validity)
936

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

941
    """
942
    (name, ) = params
943
    return backend.RemoveX509Certificate(name)
944

    
945
  # Import and export
946

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

951
    """
952
    (opts_s, instance, component, (dest, dest_args)) = params
953

    
954
    opts = objects.ImportExportOptions.FromDict(opts_s)
955

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

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

967
    """
968
    (opts_s, host, port, instance, component, (source, source_args)) = params
969

    
970
    opts = objects.ImportExportOptions.FromDict(opts_s)
971

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

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

983
    """
984
    return backend.GetImportExportStatus(params[0])
985

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

990
    """
991
    return backend.AbortImportExport(params[0])
992

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

997
    """
998
    return backend.CleanupImportExport(params[0])
999

    
1000

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

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

    
1017

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

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

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

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

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

    
1053

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

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

    
1064

    
1065
def Main():
1066
  """Main function for the node daemon.
1067

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

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