Statistics
| Branch: | Tag: | Revision:

root / lib / server / noded.py @ c6a9dffa

History | View | Annotate | Download (26.5 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_wipe(params):
188
    """Wipe a block device.
189

190
    """
191
    bdev_s, offset, size = params
192
    bdev = objects.Disk.FromDict(bdev_s)
193
    return backend.BlockdevWipe(bdev, offset, size)
194

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

199
    """
200
    bdev_s = params[0]
201
    bdev = objects.Disk.FromDict(bdev_s)
202
    return backend.BlockdevRemove(bdev)
203

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

208
    """
209
    devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params]
210
    return backend.BlockdevRename(devlist)
211

    
212
  @staticmethod
213
  def perspective_blockdev_assemble(params):
214
    """Assemble a block device.
215

216
    """
217
    bdev_s, owner, on_primary = params
218
    bdev = objects.Disk.FromDict(bdev_s)
219
    if bdev is None:
220
      raise ValueError("can't unserialize data!")
221
    return backend.BlockdevAssemble(bdev, owner, on_primary)
222

    
223
  @staticmethod
224
  def perspective_blockdev_shutdown(params):
225
    """Shutdown a block device.
226

227
    """
228
    bdev_s = params[0]
229
    bdev = objects.Disk.FromDict(bdev_s)
230
    if bdev is None:
231
      raise ValueError("can't unserialize data!")
232
    return backend.BlockdevShutdown(bdev)
233

    
234
  @staticmethod
235
  def perspective_blockdev_addchildren(params):
236
    """Add a child to a mirror device.
237

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

241
    """
242
    bdev_s, ndev_s = params
243
    bdev = objects.Disk.FromDict(bdev_s)
244
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
245
    if bdev is None or ndevs.count(None) > 0:
246
      raise ValueError("can't unserialize data!")
247
    return backend.BlockdevAddchildren(bdev, ndevs)
248

    
249
  @staticmethod
250
  def perspective_blockdev_removechildren(params):
251
    """Remove a child from a mirror device.
252

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

256
    """
257
    bdev_s, ndev_s = params
258
    bdev = objects.Disk.FromDict(bdev_s)
259
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
260
    if bdev is None or ndevs.count(None) > 0:
261
      raise ValueError("can't unserialize data!")
262
    return backend.BlockdevRemovechildren(bdev, ndevs)
263

    
264
  @staticmethod
265
  def perspective_blockdev_getmirrorstatus(params):
266
    """Return the mirror status for a list of disks.
267

268
    """
269
    disks = [objects.Disk.FromDict(dsk_s)
270
             for dsk_s in params]
271
    return [status.ToDict()
272
            for status in backend.BlockdevGetmirrorstatus(disks)]
273

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

278
    """
279
    (node_disks, ) = params
280

    
281
    node_name = netutils.Hostname.GetSysName()
282

    
283
    disks = [objects.Disk.FromDict(dsk_s)
284
             for dsk_s in node_disks.get(node_name, [])]
285

    
286
    result = []
287

    
288
    for (success, status) in backend.BlockdevGetmirrorstatusMulti(disks):
289
      if success:
290
        result.append((success, status.ToDict()))
291
      else:
292
        result.append((success, status))
293

    
294
    return result
295

    
296
  @staticmethod
297
  def perspective_blockdev_find(params):
298
    """Expose the FindBlockDevice functionality for a disk.
299

300
    This will try to find but not activate a disk.
301

302
    """
303
    disk = objects.Disk.FromDict(params[0])
304

    
305
    result = backend.BlockdevFind(disk)
306
    if result is None:
307
      return None
308

    
309
    return result.ToDict()
310

    
311
  @staticmethod
312
  def perspective_blockdev_snapshot(params):
313
    """Create a snapshot device.
314

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

319
    """
320
    cfbd = objects.Disk.FromDict(params[0])
321
    return backend.BlockdevSnapshot(cfbd)
322

    
323
  @staticmethod
324
  def perspective_blockdev_grow(params):
325
    """Grow a stack of devices.
326

327
    """
328
    cfbd = objects.Disk.FromDict(params[0])
329
    amount = params[1]
330
    return backend.BlockdevGrow(cfbd, amount)
331

    
332
  @staticmethod
333
  def perspective_blockdev_close(params):
334
    """Closes the given block devices.
335

336
    """
337
    disks = [objects.Disk.FromDict(cf) for cf in params[1]]
338
    return backend.BlockdevClose(params[0], disks)
339

    
340
  @staticmethod
341
  def perspective_blockdev_getsize(params):
342
    """Compute the sizes of the given block devices.
343

344
    """
345
    disks = [objects.Disk.FromDict(cf) for cf in params[0]]
346
    return backend.BlockdevGetsize(disks)
347

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

352
    """
353
    disk = objects.Disk.FromDict(params[0])
354
    dest_node, dest_path, cluster_name = params[1:]
355
    return backend.BlockdevExport(disk, dest_node, dest_path, cluster_name)
356

    
357
  # blockdev/drbd specific methods ----------
358

    
359
  @staticmethod
360
  def perspective_drbd_disconnect_net(params):
361
    """Disconnects the network connection of drbd disks.
362

363
    Note that this is only valid for drbd disks, so the members of the
364
    disk list must all be drbd devices.
365

366
    """
367
    nodes_ip, disks = params
368
    disks = [objects.Disk.FromDict(cf) for cf in disks]
369
    return backend.DrbdDisconnectNet(nodes_ip, disks)
370

    
371
  @staticmethod
372
  def perspective_drbd_attach_net(params):
373
    """Attaches the network connection of drbd disks.
374

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

378
    """
379
    nodes_ip, disks, instance_name, multimaster = params
380
    disks = [objects.Disk.FromDict(cf) for cf in disks]
381
    return backend.DrbdAttachNet(nodes_ip, disks,
382
                                     instance_name, multimaster)
383

    
384
  @staticmethod
385
  def perspective_drbd_wait_sync(params):
386
    """Wait until DRBD disks are synched.
387

388
    Note that this is only valid for drbd disks, so the members of the
389
    disk list must all be drbd devices.
390

391
    """
392
    nodes_ip, disks = params
393
    disks = [objects.Disk.FromDict(cf) for cf in disks]
394
    return backend.DrbdWaitSync(nodes_ip, disks)
395

    
396
  @staticmethod
397
  def perspective_drbd_helper(params):
398
    """Query drbd helper.
399

400
    """
401
    return backend.GetDrbdUsermodeHelper()
402

    
403
  # export/import  --------------------------
404

    
405
  @staticmethod
406
  def perspective_finalize_export(params):
407
    """Expose the finalize export functionality.
408

409
    """
410
    instance = objects.Instance.FromDict(params[0])
411

    
412
    snap_disks = []
413
    for disk in params[1]:
414
      if isinstance(disk, bool):
415
        snap_disks.append(disk)
416
      else:
417
        snap_disks.append(objects.Disk.FromDict(disk))
418

    
419
    return backend.FinalizeExport(instance, snap_disks)
420

    
421
  @staticmethod
422
  def perspective_export_info(params):
423
    """Query information about an existing export on this node.
424

425
    The given path may not contain an export, in which case we return
426
    None.
427

428
    """
429
    path = params[0]
430
    return backend.ExportInfo(path)
431

    
432
  @staticmethod
433
  def perspective_export_list(params):
434
    """List the available exports on this node.
435

436
    Note that as opposed to export_info, which may query data about an
437
    export in any path, this only queries the standard Ganeti path
438
    (constants.EXPORT_DIR).
439

440
    """
441
    return backend.ListExports()
442

    
443
  @staticmethod
444
  def perspective_export_remove(params):
445
    """Remove an export.
446

447
    """
448
    export = params[0]
449
    return backend.RemoveExport(export)
450

    
451
  # volume  --------------------------
452

    
453
  @staticmethod
454
  def perspective_lv_list(params):
455
    """Query the list of logical volumes in a given volume group.
456

457
    """
458
    vgname = params[0]
459
    return backend.GetVolumeList(vgname)
460

    
461
  @staticmethod
462
  def perspective_vg_list(params):
463
    """Query the list of volume groups.
464

465
    """
466
    return backend.ListVolumeGroups()
467

    
468
  # Storage --------------------------
469

    
470
  @staticmethod
471
  def perspective_storage_list(params):
472
    """Get list of storage units.
473

474
    """
475
    (su_name, su_args, name, fields) = params
476
    return storage.GetStorage(su_name, *su_args).List(name, fields)
477

    
478
  @staticmethod
479
  def perspective_storage_modify(params):
480
    """Modify a storage unit.
481

482
    """
483
    (su_name, su_args, name, changes) = params
484
    return storage.GetStorage(su_name, *su_args).Modify(name, changes)
485

    
486
  @staticmethod
487
  def perspective_storage_execute(params):
488
    """Execute an operation on a storage unit.
489

490
    """
491
    (su_name, su_args, name, op) = params
492
    return storage.GetStorage(su_name, *su_args).Execute(name, op)
493

    
494
  # bridge  --------------------------
495

    
496
  @staticmethod
497
  def perspective_bridges_exist(params):
498
    """Check if all bridges given exist on this node.
499

500
    """
501
    bridges_list = params[0]
502
    return backend.BridgesExist(bridges_list)
503

    
504
  # instance  --------------------------
505

    
506
  @staticmethod
507
  def perspective_instance_os_add(params):
508
    """Install an OS on a given instance.
509

510
    """
511
    inst_s = params[0]
512
    inst = objects.Instance.FromDict(inst_s)
513
    reinstall = params[1]
514
    debug = params[2]
515
    return backend.InstanceOsAdd(inst, reinstall, debug)
516

    
517
  @staticmethod
518
  def perspective_instance_run_rename(params):
519
    """Runs the OS rename script for an instance.
520

521
    """
522
    inst_s, old_name, debug = params
523
    inst = objects.Instance.FromDict(inst_s)
524
    return backend.RunRenameInstance(inst, old_name, debug)
525

    
526
  @staticmethod
527
  def perspective_instance_shutdown(params):
528
    """Shutdown an instance.
529

530
    """
531
    instance = objects.Instance.FromDict(params[0])
532
    timeout = params[1]
533
    return backend.InstanceShutdown(instance, timeout)
534

    
535
  @staticmethod
536
  def perspective_instance_start(params):
537
    """Start an instance.
538

539
    """
540
    instance = objects.Instance.FromDict(params[0])
541
    return backend.StartInstance(instance)
542

    
543
  @staticmethod
544
  def perspective_migration_info(params):
545
    """Gather information about an instance to be migrated.
546

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

    
551
  @staticmethod
552
  def perspective_accept_instance(params):
553
    """Prepare the node to accept an instance.
554

555
    """
556
    instance, info, target = params
557
    instance = objects.Instance.FromDict(instance)
558
    return backend.AcceptInstance(instance, info, target)
559

    
560
  @staticmethod
561
  def perspective_finalize_migration(params):
562
    """Finalize the instance migration.
563

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

    
569
  @staticmethod
570
  def perspective_instance_migrate(params):
571
    """Migrates an instance.
572

573
    """
574
    instance, target, live = params
575
    instance = objects.Instance.FromDict(instance)
576
    return backend.MigrateInstance(instance, target, live)
577

    
578
  @staticmethod
579
  def perspective_instance_reboot(params):
580
    """Reboot an instance.
581

582
    """
583
    instance = objects.Instance.FromDict(params[0])
584
    reboot_type = params[1]
585
    shutdown_timeout = params[2]
586
    return backend.InstanceReboot(instance, reboot_type, shutdown_timeout)
587

    
588
  @staticmethod
589
  def perspective_instance_info(params):
590
    """Query instance information.
591

592
    """
593
    return backend.GetInstanceInfo(params[0], params[1])
594

    
595
  @staticmethod
596
  def perspective_instance_migratable(params):
597
    """Query whether the specified instance can be migrated.
598

599
    """
600
    instance = objects.Instance.FromDict(params[0])
601
    return backend.GetInstanceMigratable(instance)
602

    
603
  @staticmethod
604
  def perspective_all_instances_info(params):
605
    """Query information about all instances.
606

607
    """
608
    return backend.GetAllInstancesInfo(params[0])
609

    
610
  @staticmethod
611
  def perspective_instance_list(params):
612
    """Query the list of running instances.
613

614
    """
615
    return backend.GetInstanceList(params[0])
616

    
617
  # node --------------------------
618

    
619
  @staticmethod
620
  def perspective_node_tcp_ping(params):
621
    """Do a TcpPing on the remote node.
622

623
    """
624
    return netutils.TcpPing(params[1], params[2], timeout=params[3],
625
                            live_port_needed=params[4], source=params[0])
626

    
627
  @staticmethod
628
  def perspective_node_has_ip_address(params):
629
    """Checks if a node has the given ip address.
630

631
    """
632
    return netutils.IPAddress.Own(params[0])
633

    
634
  @staticmethod
635
  def perspective_node_info(params):
636
    """Query node information.
637

638
    """
639
    vgname, hypervisor_type = params
640
    return backend.GetNodeInfo(vgname, hypervisor_type)
641

    
642
  @staticmethod
643
  def perspective_etc_hosts_modify(params):
644
    """Modify a node entry in /etc/hosts.
645

646
    """
647
    backend.EtcHostsModify(params[0], params[1], params[2])
648

    
649
    return True
650

    
651
  @staticmethod
652
  def perspective_node_verify(params):
653
    """Run a verify sequence on this node.
654

655
    """
656
    return backend.VerifyNode(params[0], params[1])
657

    
658
  @staticmethod
659
  def perspective_node_start_master(params):
660
    """Promote this node to master status.
661

662
    """
663
    return backend.StartMaster(params[0], params[1])
664

    
665
  @staticmethod
666
  def perspective_node_stop_master(params):
667
    """Demote this node from master status.
668

669
    """
670
    return backend.StopMaster(params[0])
671

    
672
  @staticmethod
673
  def perspective_node_leave_cluster(params):
674
    """Cleanup after leaving a cluster.
675

676
    """
677
    return backend.LeaveCluster(params[0])
678

    
679
  @staticmethod
680
  def perspective_node_volumes(params):
681
    """Query the list of all logical volume groups.
682

683
    """
684
    return backend.NodeVolumes()
685

    
686
  @staticmethod
687
  def perspective_node_demote_from_mc(params):
688
    """Demote a node from the master candidate role.
689

690
    """
691
    return backend.DemoteFromMC()
692

    
693

    
694
  @staticmethod
695
  def perspective_node_powercycle(params):
696
    """Tries to powercycle the nod.
697

698
    """
699
    hypervisor_type = params[0]
700
    return backend.PowercycleNode(hypervisor_type)
701

    
702

    
703
  # cluster --------------------------
704

    
705
  @staticmethod
706
  def perspective_version(params):
707
    """Query version information.
708

709
    """
710
    return constants.PROTOCOL_VERSION
711

    
712
  @staticmethod
713
  def perspective_upload_file(params):
714
    """Upload a file.
715

716
    Note that the backend implementation imposes strict rules on which
717
    files are accepted.
718

719
    """
720
    return backend.UploadFile(*params)
721

    
722
  @staticmethod
723
  def perspective_master_info(params):
724
    """Query master information.
725

726
    """
727
    return backend.GetMasterInfo()
728

    
729
  @staticmethod
730
  def perspective_write_ssconf_files(params):
731
    """Write ssconf files.
732

733
    """
734
    (values,) = params
735
    return backend.WriteSsconfFiles(values)
736

    
737
  # os -----------------------
738

    
739
  @staticmethod
740
  def perspective_os_diagnose(params):
741
    """Query detailed information about existing OSes.
742

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

    
746
  @staticmethod
747
  def perspective_os_get(params):
748
    """Query information about a given OS.
749

750
    """
751
    name = params[0]
752
    os_obj = backend.OSFromDisk(name)
753
    return os_obj.ToDict()
754

    
755
  @staticmethod
756
  def perspective_os_validate(params):
757
    """Run a given OS' validation routine.
758

759
    """
760
    required, name, checks, params = params
761
    return backend.ValidateOS(required, name, checks, params)
762

    
763
  # hooks -----------------------
764

    
765
  @staticmethod
766
  def perspective_hooks_runner(params):
767
    """Run hook scripts.
768

769
    """
770
    hpath, phase, env = params
771
    hr = backend.HooksRunner()
772
    return hr.RunHooks(hpath, phase, env)
773

    
774
  # iallocator -----------------
775

    
776
  @staticmethod
777
  def perspective_iallocator_runner(params):
778
    """Run an iallocator script.
779

780
    """
781
    name, idata = params
782
    iar = backend.IAllocatorRunner()
783
    return iar.Run(name, idata)
784

    
785
  # test -----------------------
786

    
787
  @staticmethod
788
  def perspective_test_delay(params):
789
    """Run test delay.
790

791
    """
792
    duration = params[0]
793
    status, rval = utils.TestDelay(duration)
794
    if not status:
795
      raise backend.RPCFail(rval)
796
    return rval
797

    
798
  # file storage ---------------
799

    
800
  @staticmethod
801
  def perspective_file_storage_dir_create(params):
802
    """Create the file storage directory.
803

804
    """
805
    file_storage_dir = params[0]
806
    return backend.CreateFileStorageDir(file_storage_dir)
807

    
808
  @staticmethod
809
  def perspective_file_storage_dir_remove(params):
810
    """Remove the file storage directory.
811

812
    """
813
    file_storage_dir = params[0]
814
    return backend.RemoveFileStorageDir(file_storage_dir)
815

    
816
  @staticmethod
817
  def perspective_file_storage_dir_rename(params):
818
    """Rename the file storage directory.
819

820
    """
821
    old_file_storage_dir = params[0]
822
    new_file_storage_dir = params[1]
823
    return backend.RenameFileStorageDir(old_file_storage_dir,
824
                                        new_file_storage_dir)
825

    
826
  # jobs ------------------------
827

    
828
  @staticmethod
829
  @_RequireJobQueueLock
830
  def perspective_jobqueue_update(params):
831
    """Update job queue.
832

833
    """
834
    (file_name, content) = params
835
    return backend.JobQueueUpdate(file_name, content)
836

    
837
  @staticmethod
838
  @_RequireJobQueueLock
839
  def perspective_jobqueue_purge(params):
840
    """Purge job queue.
841

842
    """
843
    return backend.JobQueuePurge()
844

    
845
  @staticmethod
846
  @_RequireJobQueueLock
847
  def perspective_jobqueue_rename(params):
848
    """Rename a job queue file.
849

850
    """
851
    # TODO: What if a file fails to rename?
852
    return [backend.JobQueueRename(old, new) for old, new in params]
853

    
854
  # hypervisor ---------------
855

    
856
  @staticmethod
857
  def perspective_hypervisor_validate_params(params):
858
    """Validate the hypervisor parameters.
859

860
    """
861
    (hvname, hvparams) = params
862
    return backend.ValidateHVParams(hvname, hvparams)
863

    
864
  # Crypto
865

    
866
  @staticmethod
867
  def perspective_x509_cert_create(params):
868
    """Creates a new X509 certificate for SSL/TLS.
869

870
    """
871
    (validity, ) = params
872
    return backend.CreateX509Certificate(validity)
873

    
874
  @staticmethod
875
  def perspective_x509_cert_remove(params):
876
    """Removes a X509 certificate.
877

878
    """
879
    (name, ) = params
880
    return backend.RemoveX509Certificate(name)
881

    
882
  # Import and export
883

    
884
  @staticmethod
885
  def perspective_import_start(params):
886
    """Starts an import daemon.
887

888
    """
889
    (opts_s, instance, dest, dest_args) = params
890

    
891
    opts = objects.ImportExportOptions.FromDict(opts_s)
892

    
893
    return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
894
                                           None, None,
895
                                           objects.Instance.FromDict(instance),
896
                                           dest,
897
                                           _DecodeImportExportIO(dest,
898
                                                                 dest_args))
899

    
900
  @staticmethod
901
  def perspective_export_start(params):
902
    """Starts an export daemon.
903

904
    """
905
    (opts_s, host, port, instance, source, source_args) = params
906

    
907
    opts = objects.ImportExportOptions.FromDict(opts_s)
908

    
909
    return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
910
                                           host, port,
911
                                           objects.Instance.FromDict(instance),
912
                                           source,
913
                                           _DecodeImportExportIO(source,
914
                                                                 source_args))
915

    
916
  @staticmethod
917
  def perspective_impexp_status(params):
918
    """Retrieves the status of an import or export daemon.
919

920
    """
921
    return backend.GetImportExportStatus(params[0])
922

    
923
  @staticmethod
924
  def perspective_impexp_abort(params):
925
    """Aborts an import or export.
926

927
    """
928
    return backend.AbortImportExport(params[0])
929

    
930
  @staticmethod
931
  def perspective_impexp_cleanup(params):
932
    """Cleans up after an import or export.
933

934
    """
935
    return backend.CleanupImportExport(params[0])
936

    
937

    
938
def CheckNoded(_, args):
939
  """Initial checks whether to run or exit with a failure.
940

941
  """
942
  if args: # noded doesn't take any arguments
943
    print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
944
                          sys.argv[0])
945
    sys.exit(constants.EXIT_FAILURE)
946

    
947

    
948
def PrepNoded(options, _):
949
  """Preparation node daemon function, executed with the PID file held.
950

951
  """
952
  if options.mlock:
953
    request_executor_class = MlockallRequestExecutor
954
    try:
955
      utils.Mlockall()
956
    except errors.NoCtypesError:
957
      logging.warning("Cannot set memory lock, ctypes module not found")
958
      request_executor_class = http.server.HttpServerRequestExecutor
959
  else:
960
    request_executor_class = http.server.HttpServerRequestExecutor
961

    
962
  # Read SSL certificate
963
  if options.ssl:
964
    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
965
                                    ssl_cert_path=options.ssl_cert)
966
  else:
967
    ssl_params = None
968

    
969
  err = _PrepareQueueLock()
970
  if err is not None:
971
    # this might be some kind of file-system/permission error; while
972
    # this breaks the job queue functionality, we shouldn't prevent
973
    # startup of the whole node daemon because of this
974
    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
975

    
976
  mainloop = daemon.Mainloop()
977
  server = NodeHttpServer(mainloop, options.bind_address, options.port,
978
                          ssl_params=ssl_params, ssl_verify_peer=True,
979
                          request_executor_class=request_executor_class)
980
  server.Start()
981
  return (mainloop, server)
982

    
983

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

987
  """
988
  (mainloop, server) = prep_data
989
  try:
990
    mainloop.Run()
991
  finally:
992
    server.Stop()
993

    
994

    
995
def Main():
996
  """Main function for the node daemon.
997

998
  """
999
  parser = OptionParser(description="Ganeti node daemon",
1000
                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]",
1001
                        version="%%prog (ganeti) %s" %
1002
                        constants.RELEASE_VERSION)
1003
  parser.add_option("--no-mlock", dest="mlock",
1004
                    help="Do not mlock the node memory in ram",
1005
                    default=True, action="store_false")
1006

    
1007
  daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded,
1008
                     default_ssl_cert=constants.NODED_CERT_FILE,
1009
                     default_ssl_key=constants.NODED_CERT_FILE,
1010
                     console_logging=True)