Statistics
| Branch: | Tag: | Revision:

root / lib / server / noded.py @ 9c007da8

History | View | Annotate | Download (27 kB)

1
#
2
#
3

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

    
37
from optparse import OptionParser
38

    
39
from ganeti import backend
40
from ganeti import constants
41
from ganeti import objects
42
from ganeti import errors
43
from ganeti import jstore
44
from ganeti import daemon
45
from ganeti import http
46
from ganeti import utils
47
from ganeti import storage
48
from ganeti import serializer
49
from ganeti import netutils
50

    
51
import ganeti.http.server # pylint: disable-msg=W0611
52

    
53

    
54
queue_lock = None
55

    
56

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

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

62
  """
63
  global queue_lock # pylint: disable-msg=W0603
64

    
65
  if queue_lock is not None:
66
    return None
67

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

    
75

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

79
  """
80
  QUEUE_LOCK_TIMEOUT = 10
81

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

    
94
  return wrapper
95

    
96

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

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

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

    
109
  return ieioargs
110

    
111

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

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

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

    
122

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

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

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

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

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

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

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

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

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

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

    
173
  # the new block devices  --------------------------
174

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
290
    node_name = netutils.Hostname.GetSysName()
291

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

    
295
    result = []
296

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

    
303
    return result
304

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

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

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

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

    
318
    return result.ToDict()
319

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

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

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

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

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

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

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

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

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

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

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

    
366
  # blockdev/drbd specific methods ----------
367

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

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

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

    
380
  @staticmethod
381
  def perspective_drbd_attach_net(params):
382
    """Attaches the network connection of drbd disks.
383

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

387
    """
388
    nodes_ip, disks, instance_name, multimaster = params
389
    disks = [objects.Disk.FromDict(cf) for cf in disks]
390
    return backend.DrbdAttachNet(nodes_ip, disks,
391
                                     instance_name, multimaster)
392

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

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

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

    
405
  @staticmethod
406
  def perspective_drbd_helper(params):
407
    """Query drbd helper.
408

409
    """
410
    return backend.GetDrbdUsermodeHelper()
411

    
412
  # export/import  --------------------------
413

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

418
    """
419
    instance = objects.Instance.FromDict(params[0])
420

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

    
428
    return backend.FinalizeExport(instance, snap_disks)
429

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

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

437
    """
438
    path = params[0]
439
    return backend.ExportInfo(path)
440

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

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

449
    """
450
    return backend.ListExports()
451

    
452
  @staticmethod
453
  def perspective_export_remove(params):
454
    """Remove an export.
455

456
    """
457
    export = params[0]
458
    return backend.RemoveExport(export)
459

    
460
  # volume  --------------------------
461

    
462
  @staticmethod
463
  def perspective_lv_list(params):
464
    """Query the list of logical volumes in a given volume group.
465

466
    """
467
    vgname = params[0]
468
    return backend.GetVolumeList(vgname)
469

    
470
  @staticmethod
471
  def perspective_vg_list(params):
472
    """Query the list of volume groups.
473

474
    """
475
    return backend.ListVolumeGroups()
476

    
477
  # Storage --------------------------
478

    
479
  @staticmethod
480
  def perspective_storage_list(params):
481
    """Get list of storage units.
482

483
    """
484
    (su_name, su_args, name, fields) = params
485
    return storage.GetStorage(su_name, *su_args).List(name, fields)
486

    
487
  @staticmethod
488
  def perspective_storage_modify(params):
489
    """Modify a storage unit.
490

491
    """
492
    (su_name, su_args, name, changes) = params
493
    return storage.GetStorage(su_name, *su_args).Modify(name, changes)
494

    
495
  @staticmethod
496
  def perspective_storage_execute(params):
497
    """Execute an operation on a storage unit.
498

499
    """
500
    (su_name, su_args, name, op) = params
501
    return storage.GetStorage(su_name, *su_args).Execute(name, op)
502

    
503
  # bridge  --------------------------
504

    
505
  @staticmethod
506
  def perspective_bridges_exist(params):
507
    """Check if all bridges given exist on this node.
508

509
    """
510
    bridges_list = params[0]
511
    return backend.BridgesExist(bridges_list)
512

    
513
  # instance  --------------------------
514

    
515
  @staticmethod
516
  def perspective_instance_os_add(params):
517
    """Install an OS on a given instance.
518

519
    """
520
    inst_s = params[0]
521
    inst = objects.Instance.FromDict(inst_s)
522
    reinstall = params[1]
523
    debug = params[2]
524
    return backend.InstanceOsAdd(inst, reinstall, debug)
525

    
526
  @staticmethod
527
  def perspective_instance_run_rename(params):
528
    """Runs the OS rename script for an instance.
529

530
    """
531
    inst_s, old_name, debug = params
532
    inst = objects.Instance.FromDict(inst_s)
533
    return backend.RunRenameInstance(inst, old_name, debug)
534

    
535
  @staticmethod
536
  def perspective_instance_shutdown(params):
537
    """Shutdown an instance.
538

539
    """
540
    instance = objects.Instance.FromDict(params[0])
541
    timeout = params[1]
542
    return backend.InstanceShutdown(instance, timeout)
543

    
544
  @staticmethod
545
  def perspective_instance_start(params):
546
    """Start an instance.
547

548
    """
549
    instance = objects.Instance.FromDict(params[0])
550
    return backend.StartInstance(instance)
551

    
552
  @staticmethod
553
  def perspective_migration_info(params):
554
    """Gather information about an instance to be migrated.
555

556
    """
557
    instance = objects.Instance.FromDict(params[0])
558
    return backend.MigrationInfo(instance)
559

    
560
  @staticmethod
561
  def perspective_accept_instance(params):
562
    """Prepare the node to accept an instance.
563

564
    """
565
    instance, info, target = params
566
    instance = objects.Instance.FromDict(instance)
567
    return backend.AcceptInstance(instance, info, target)
568

    
569
  @staticmethod
570
  def perspective_finalize_migration(params):
571
    """Finalize the instance migration.
572

573
    """
574
    instance, info, success = params
575
    instance = objects.Instance.FromDict(instance)
576
    return backend.FinalizeMigration(instance, info, success)
577

    
578
  @staticmethod
579
  def perspective_instance_migrate(params):
580
    """Migrates an instance.
581

582
    """
583
    instance, target, live = params
584
    instance = objects.Instance.FromDict(instance)
585
    return backend.MigrateInstance(instance, target, live)
586

    
587
  @staticmethod
588
  def perspective_instance_reboot(params):
589
    """Reboot an instance.
590

591
    """
592
    instance = objects.Instance.FromDict(params[0])
593
    reboot_type = params[1]
594
    shutdown_timeout = params[2]
595
    return backend.InstanceReboot(instance, reboot_type, shutdown_timeout)
596

    
597
  @staticmethod
598
  def perspective_instance_info(params):
599
    """Query instance information.
600

601
    """
602
    return backend.GetInstanceInfo(params[0], params[1])
603

    
604
  @staticmethod
605
  def perspective_instance_migratable(params):
606
    """Query whether the specified instance can be migrated.
607

608
    """
609
    instance = objects.Instance.FromDict(params[0])
610
    return backend.GetInstanceMigratable(instance)
611

    
612
  @staticmethod
613
  def perspective_all_instances_info(params):
614
    """Query information about all instances.
615

616
    """
617
    return backend.GetAllInstancesInfo(params[0])
618

    
619
  @staticmethod
620
  def perspective_instance_list(params):
621
    """Query the list of running instances.
622

623
    """
624
    return backend.GetInstanceList(params[0])
625

    
626
  # node --------------------------
627

    
628
  @staticmethod
629
  def perspective_node_tcp_ping(params):
630
    """Do a TcpPing on the remote node.
631

632
    """
633
    return netutils.TcpPing(params[1], params[2], timeout=params[3],
634
                            live_port_needed=params[4], source=params[0])
635

    
636
  @staticmethod
637
  def perspective_node_has_ip_address(params):
638
    """Checks if a node has the given ip address.
639

640
    """
641
    return netutils.IPAddress.Own(params[0])
642

    
643
  @staticmethod
644
  def perspective_node_info(params):
645
    """Query node information.
646

647
    """
648
    vgname, hypervisor_type = params
649
    return backend.GetNodeInfo(vgname, hypervisor_type)
650

    
651
  @staticmethod
652
  def perspective_etc_hosts_modify(params):
653
    """Modify a node entry in /etc/hosts.
654

655
    """
656
    backend.EtcHostsModify(params[0], params[1], params[2])
657

    
658
    return True
659

    
660
  @staticmethod
661
  def perspective_node_verify(params):
662
    """Run a verify sequence on this node.
663

664
    """
665
    return backend.VerifyNode(params[0], params[1])
666

    
667
  @staticmethod
668
  def perspective_node_start_master(params):
669
    """Promote this node to master status.
670

671
    """
672
    return backend.StartMaster(params[0], params[1])
673

    
674
  @staticmethod
675
  def perspective_node_stop_master(params):
676
    """Demote this node from master status.
677

678
    """
679
    return backend.StopMaster(params[0])
680

    
681
  @staticmethod
682
  def perspective_node_leave_cluster(params):
683
    """Cleanup after leaving a cluster.
684

685
    """
686
    return backend.LeaveCluster(params[0])
687

    
688
  @staticmethod
689
  def perspective_node_volumes(params):
690
    """Query the list of all logical volume groups.
691

692
    """
693
    return backend.NodeVolumes()
694

    
695
  @staticmethod
696
  def perspective_node_demote_from_mc(params):
697
    """Demote a node from the master candidate role.
698

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

    
702

    
703
  @staticmethod
704
  def perspective_node_powercycle(params):
705
    """Tries to powercycle the nod.
706

707
    """
708
    hypervisor_type = params[0]
709
    return backend.PowercycleNode(hypervisor_type)
710

    
711

    
712
  # cluster --------------------------
713

    
714
  @staticmethod
715
  def perspective_version(params):
716
    """Query version information.
717

718
    """
719
    return constants.PROTOCOL_VERSION
720

    
721
  @staticmethod
722
  def perspective_upload_file(params):
723
    """Upload a file.
724

725
    Note that the backend implementation imposes strict rules on which
726
    files are accepted.
727

728
    """
729
    return backend.UploadFile(*params)
730

    
731
  @staticmethod
732
  def perspective_master_info(params):
733
    """Query master information.
734

735
    """
736
    return backend.GetMasterInfo()
737

    
738
  @staticmethod
739
  def perspective_run_oob(params):
740
    """Runs oob on node.
741

742
    """
743
    output = backend.RunOob(params[0], params[1], params[2], params[3])
744
    if output:
745
      result = serializer.LoadJson(output)
746
    else:
747
      result = None
748
    return result
749

    
750
  @staticmethod
751
  def perspective_write_ssconf_files(params):
752
    """Write ssconf files.
753

754
    """
755
    (values,) = params
756
    return backend.WriteSsconfFiles(values)
757

    
758
  # os -----------------------
759

    
760
  @staticmethod
761
  def perspective_os_diagnose(params):
762
    """Query detailed information about existing OSes.
763

764
    """
765
    return backend.DiagnoseOS()
766

    
767
  @staticmethod
768
  def perspective_os_get(params):
769
    """Query information about a given OS.
770

771
    """
772
    name = params[0]
773
    os_obj = backend.OSFromDisk(name)
774
    return os_obj.ToDict()
775

    
776
  @staticmethod
777
  def perspective_os_validate(params):
778
    """Run a given OS' validation routine.
779

780
    """
781
    required, name, checks, params = params
782
    return backend.ValidateOS(required, name, checks, params)
783

    
784
  # hooks -----------------------
785

    
786
  @staticmethod
787
  def perspective_hooks_runner(params):
788
    """Run hook scripts.
789

790
    """
791
    hpath, phase, env = params
792
    hr = backend.HooksRunner()
793
    return hr.RunHooks(hpath, phase, env)
794

    
795
  # iallocator -----------------
796

    
797
  @staticmethod
798
  def perspective_iallocator_runner(params):
799
    """Run an iallocator script.
800

801
    """
802
    name, idata = params
803
    iar = backend.IAllocatorRunner()
804
    return iar.Run(name, idata)
805

    
806
  # test -----------------------
807

    
808
  @staticmethod
809
  def perspective_test_delay(params):
810
    """Run test delay.
811

812
    """
813
    duration = params[0]
814
    status, rval = utils.TestDelay(duration)
815
    if not status:
816
      raise backend.RPCFail(rval)
817
    return rval
818

    
819
  # file storage ---------------
820

    
821
  @staticmethod
822
  def perspective_file_storage_dir_create(params):
823
    """Create the file storage directory.
824

825
    """
826
    file_storage_dir = params[0]
827
    return backend.CreateFileStorageDir(file_storage_dir)
828

    
829
  @staticmethod
830
  def perspective_file_storage_dir_remove(params):
831
    """Remove the file storage directory.
832

833
    """
834
    file_storage_dir = params[0]
835
    return backend.RemoveFileStorageDir(file_storage_dir)
836

    
837
  @staticmethod
838
  def perspective_file_storage_dir_rename(params):
839
    """Rename the file storage directory.
840

841
    """
842
    old_file_storage_dir = params[0]
843
    new_file_storage_dir = params[1]
844
    return backend.RenameFileStorageDir(old_file_storage_dir,
845
                                        new_file_storage_dir)
846

    
847
  # jobs ------------------------
848

    
849
  @staticmethod
850
  @_RequireJobQueueLock
851
  def perspective_jobqueue_update(params):
852
    """Update job queue.
853

854
    """
855
    (file_name, content) = params
856
    return backend.JobQueueUpdate(file_name, content)
857

    
858
  @staticmethod
859
  @_RequireJobQueueLock
860
  def perspective_jobqueue_purge(params):
861
    """Purge job queue.
862

863
    """
864
    return backend.JobQueuePurge()
865

    
866
  @staticmethod
867
  @_RequireJobQueueLock
868
  def perspective_jobqueue_rename(params):
869
    """Rename a job queue file.
870

871
    """
872
    # TODO: What if a file fails to rename?
873
    return [backend.JobQueueRename(old, new) for old, new in params]
874

    
875
  # hypervisor ---------------
876

    
877
  @staticmethod
878
  def perspective_hypervisor_validate_params(params):
879
    """Validate the hypervisor parameters.
880

881
    """
882
    (hvname, hvparams) = params
883
    return backend.ValidateHVParams(hvname, hvparams)
884

    
885
  # Crypto
886

    
887
  @staticmethod
888
  def perspective_x509_cert_create(params):
889
    """Creates a new X509 certificate for SSL/TLS.
890

891
    """
892
    (validity, ) = params
893
    return backend.CreateX509Certificate(validity)
894

    
895
  @staticmethod
896
  def perspective_x509_cert_remove(params):
897
    """Removes a X509 certificate.
898

899
    """
900
    (name, ) = params
901
    return backend.RemoveX509Certificate(name)
902

    
903
  # Import and export
904

    
905
  @staticmethod
906
  def perspective_import_start(params):
907
    """Starts an import daemon.
908

909
    """
910
    (opts_s, instance, dest, dest_args) = params
911

    
912
    opts = objects.ImportExportOptions.FromDict(opts_s)
913

    
914
    return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
915
                                           None, None,
916
                                           objects.Instance.FromDict(instance),
917
                                           dest,
918
                                           _DecodeImportExportIO(dest,
919
                                                                 dest_args))
920

    
921
  @staticmethod
922
  def perspective_export_start(params):
923
    """Starts an export daemon.
924

925
    """
926
    (opts_s, host, port, instance, source, source_args) = params
927

    
928
    opts = objects.ImportExportOptions.FromDict(opts_s)
929

    
930
    return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
931
                                           host, port,
932
                                           objects.Instance.FromDict(instance),
933
                                           source,
934
                                           _DecodeImportExportIO(source,
935
                                                                 source_args))
936

    
937
  @staticmethod
938
  def perspective_impexp_status(params):
939
    """Retrieves the status of an import or export daemon.
940

941
    """
942
    return backend.GetImportExportStatus(params[0])
943

    
944
  @staticmethod
945
  def perspective_impexp_abort(params):
946
    """Aborts an import or export.
947

948
    """
949
    return backend.AbortImportExport(params[0])
950

    
951
  @staticmethod
952
  def perspective_impexp_cleanup(params):
953
    """Cleans up after an import or export.
954

955
    """
956
    return backend.CleanupImportExport(params[0])
957

    
958

    
959
def CheckNoded(_, args):
960
  """Initial checks whether to run or exit with a failure.
961

962
  """
963
  if args: # noded doesn't take any arguments
964
    print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
965
                          sys.argv[0])
966
    sys.exit(constants.EXIT_FAILURE)
967

    
968

    
969
def PrepNoded(options, _):
970
  """Preparation node daemon function, executed with the PID file held.
971

972
  """
973
  if options.mlock:
974
    request_executor_class = MlockallRequestExecutor
975
    try:
976
      utils.Mlockall()
977
    except errors.NoCtypesError:
978
      logging.warning("Cannot set memory lock, ctypes module not found")
979
      request_executor_class = http.server.HttpServerRequestExecutor
980
  else:
981
    request_executor_class = http.server.HttpServerRequestExecutor
982

    
983
  # Read SSL certificate
984
  if options.ssl:
985
    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
986
                                    ssl_cert_path=options.ssl_cert)
987
  else:
988
    ssl_params = None
989

    
990
  err = _PrepareQueueLock()
991
  if err is not None:
992
    # this might be some kind of file-system/permission error; while
993
    # this breaks the job queue functionality, we shouldn't prevent
994
    # startup of the whole node daemon because of this
995
    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
996

    
997
  mainloop = daemon.Mainloop()
998
  server = NodeHttpServer(mainloop, options.bind_address, options.port,
999
                          ssl_params=ssl_params, ssl_verify_peer=True,
1000
                          request_executor_class=request_executor_class)
1001
  server.Start()
1002
  return (mainloop, server)
1003

    
1004

    
1005
def ExecNoded(options, args, prep_data): # pylint: disable-msg=W0613
1006
  """Main node daemon function, executed with the PID file held.
1007

1008
  """
1009
  (mainloop, server) = prep_data
1010
  try:
1011
    mainloop.Run()
1012
  finally:
1013
    server.Stop()
1014

    
1015

    
1016
def Main():
1017
  """Main function for the node daemon.
1018

1019
  """
1020
  parser = OptionParser(description="Ganeti node daemon",
1021
                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]",
1022
                        version="%%prog (ganeti) %s" %
1023
                        constants.RELEASE_VERSION)
1024
  parser.add_option("--no-mlock", dest="mlock",
1025
                    help="Do not mlock the node memory in ram",
1026
                    default=True, action="store_false")
1027

    
1028
  daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded,
1029
                     default_ssl_cert=constants.NODED_CERT_FILE,
1030
                     default_ssl_key=constants.NODED_CERT_FILE,
1031
                     console_logging=True)