Statistics
| Branch: | Tag: | Revision:

root / daemons / ganeti-noded @ 3ee53f1f

History | View | Annotate | Download (25.8 kB)

1
#!/usr/bin/python
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_remove(params):
188
    """Remove a block device.
189

    
190
    """
191
    bdev_s = params[0]
192
    bdev = objects.Disk.FromDict(bdev_s)
193
    return backend.BlockdevRemove(bdev)
194

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

    
199
    """
200
    devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params]
201
    return backend.BlockdevRename(devlist)
202

    
203
  @staticmethod
204
  def perspective_blockdev_assemble(params):
205
    """Assemble a block device.
206

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

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

    
218
    """
219
    bdev_s = params[0]
220
    bdev = objects.Disk.FromDict(bdev_s)
221
    if bdev is None:
222
      raise ValueError("can't unserialize data!")
223
    return backend.BlockdevShutdown(bdev)
224

    
225
  @staticmethod
226
  def perspective_blockdev_addchildren(params):
227
    """Add a child to a mirror device.
228

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

    
232
    """
233
    bdev_s, ndev_s = params
234
    bdev = objects.Disk.FromDict(bdev_s)
235
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
236
    if bdev is None or ndevs.count(None) > 0:
237
      raise ValueError("can't unserialize data!")
238
    return backend.BlockdevAddchildren(bdev, ndevs)
239

    
240
  @staticmethod
241
  def perspective_blockdev_removechildren(params):
242
    """Remove a child from a mirror device.
243

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

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

    
255
  @staticmethod
256
  def perspective_blockdev_getmirrorstatus(params):
257
    """Return the mirror status for a list of disks.
258

    
259
    """
260
    disks = [objects.Disk.FromDict(dsk_s)
261
             for dsk_s in params]
262
    return [status.ToDict()
263
            for status in backend.BlockdevGetmirrorstatus(disks)]
264

    
265
  @staticmethod
266
  def perspective_blockdev_find(params):
267
    """Expose the FindBlockDevice functionality for a disk.
268

    
269
    This will try to find but not activate a disk.
270

    
271
    """
272
    disk = objects.Disk.FromDict(params[0])
273

    
274
    result = backend.BlockdevFind(disk)
275
    if result is None:
276
      return None
277

    
278
    return result.ToDict()
279

    
280
  @staticmethod
281
  def perspective_blockdev_snapshot(params):
282
    """Create a snapshot device.
283

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

    
288
    """
289
    cfbd = objects.Disk.FromDict(params[0])
290
    return backend.BlockdevSnapshot(cfbd)
291

    
292
  @staticmethod
293
  def perspective_blockdev_grow(params):
294
    """Grow a stack of devices.
295

    
296
    """
297
    cfbd = objects.Disk.FromDict(params[0])
298
    amount = params[1]
299
    return backend.BlockdevGrow(cfbd, amount)
300

    
301
  @staticmethod
302
  def perspective_blockdev_close(params):
303
    """Closes the given block devices.
304

    
305
    """
306
    disks = [objects.Disk.FromDict(cf) for cf in params[1]]
307
    return backend.BlockdevClose(params[0], disks)
308

    
309
  @staticmethod
310
  def perspective_blockdev_getsize(params):
311
    """Compute the sizes of the given block devices.
312

    
313
    """
314
    disks = [objects.Disk.FromDict(cf) for cf in params[0]]
315
    return backend.BlockdevGetsize(disks)
316

    
317
  @staticmethod
318
  def perspective_blockdev_export(params):
319
    """Compute the sizes of the given block devices.
320

    
321
    """
322
    disk = objects.Disk.FromDict(params[0])
323
    dest_node, dest_path, cluster_name = params[1:]
324
    return backend.BlockdevExport(disk, dest_node, dest_path, cluster_name)
325

    
326
  # blockdev/drbd specific methods ----------
327

    
328
  @staticmethod
329
  def perspective_drbd_disconnect_net(params):
330
    """Disconnects the network connection of drbd disks.
331

    
332
    Note that this is only valid for drbd disks, so the members of the
333
    disk list must all be drbd devices.
334

    
335
    """
336
    nodes_ip, disks = params
337
    disks = [objects.Disk.FromDict(cf) for cf in disks]
338
    return backend.DrbdDisconnectNet(nodes_ip, disks)
339

    
340
  @staticmethod
341
  def perspective_drbd_attach_net(params):
342
    """Attaches the network connection of drbd disks.
343

    
344
    Note that this is only valid for drbd disks, so the members of the
345
    disk list must all be drbd devices.
346

    
347
    """
348
    nodes_ip, disks, instance_name, multimaster = params
349
    disks = [objects.Disk.FromDict(cf) for cf in disks]
350
    return backend.DrbdAttachNet(nodes_ip, disks,
351
                                     instance_name, multimaster)
352

    
353
  @staticmethod
354
  def perspective_drbd_wait_sync(params):
355
    """Wait until DRBD disks are synched.
356

    
357
    Note that this is only valid for drbd disks, so the members of the
358
    disk list must all be drbd devices.
359

    
360
    """
361
    nodes_ip, disks = params
362
    disks = [objects.Disk.FromDict(cf) for cf in disks]
363
    return backend.DrbdWaitSync(nodes_ip, disks)
364

    
365
  @staticmethod
366
  def perspective_drbd_helper(params):
367
    """Query drbd helper.
368

    
369
    """
370
    return backend.GetDrbdUsermodeHelper()
371

    
372
  # export/import  --------------------------
373

    
374
  @staticmethod
375
  def perspective_finalize_export(params):
376
    """Expose the finalize export functionality.
377

    
378
    """
379
    instance = objects.Instance.FromDict(params[0])
380

    
381
    snap_disks = []
382
    for disk in params[1]:
383
      if isinstance(disk, bool):
384
        snap_disks.append(disk)
385
      else:
386
        snap_disks.append(objects.Disk.FromDict(disk))
387

    
388
    return backend.FinalizeExport(instance, snap_disks)
389

    
390
  @staticmethod
391
  def perspective_export_info(params):
392
    """Query information about an existing export on this node.
393

    
394
    The given path may not contain an export, in which case we return
395
    None.
396

    
397
    """
398
    path = params[0]
399
    return backend.ExportInfo(path)
400

    
401
  @staticmethod
402
  def perspective_export_list(params):
403
    """List the available exports on this node.
404

    
405
    Note that as opposed to export_info, which may query data about an
406
    export in any path, this only queries the standard Ganeti path
407
    (constants.EXPORT_DIR).
408

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

    
412
  @staticmethod
413
  def perspective_export_remove(params):
414
    """Remove an export.
415

    
416
    """
417
    export = params[0]
418
    return backend.RemoveExport(export)
419

    
420
  # volume  --------------------------
421

    
422
  @staticmethod
423
  def perspective_lv_list(params):
424
    """Query the list of logical volumes in a given volume group.
425

    
426
    """
427
    vgname = params[0]
428
    return backend.GetVolumeList(vgname)
429

    
430
  @staticmethod
431
  def perspective_vg_list(params):
432
    """Query the list of volume groups.
433

    
434
    """
435
    return backend.ListVolumeGroups()
436

    
437
  # Storage --------------------------
438

    
439
  @staticmethod
440
  def perspective_storage_list(params):
441
    """Get list of storage units.
442

    
443
    """
444
    (su_name, su_args, name, fields) = params
445
    return storage.GetStorage(su_name, *su_args).List(name, fields)
446

    
447
  @staticmethod
448
  def perspective_storage_modify(params):
449
    """Modify a storage unit.
450

    
451
    """
452
    (su_name, su_args, name, changes) = params
453
    return storage.GetStorage(su_name, *su_args).Modify(name, changes)
454

    
455
  @staticmethod
456
  def perspective_storage_execute(params):
457
    """Execute an operation on a storage unit.
458

    
459
    """
460
    (su_name, su_args, name, op) = params
461
    return storage.GetStorage(su_name, *su_args).Execute(name, op)
462

    
463
  # bridge  --------------------------
464

    
465
  @staticmethod
466
  def perspective_bridges_exist(params):
467
    """Check if all bridges given exist on this node.
468

    
469
    """
470
    bridges_list = params[0]
471
    return backend.BridgesExist(bridges_list)
472

    
473
  # instance  --------------------------
474

    
475
  @staticmethod
476
  def perspective_instance_os_add(params):
477
    """Install an OS on a given instance.
478

    
479
    """
480
    inst_s = params[0]
481
    inst = objects.Instance.FromDict(inst_s)
482
    reinstall = params[1]
483
    debug = params[2]
484
    return backend.InstanceOsAdd(inst, reinstall, debug)
485

    
486
  @staticmethod
487
  def perspective_instance_run_rename(params):
488
    """Runs the OS rename script for an instance.
489

    
490
    """
491
    inst_s, old_name, debug = params
492
    inst = objects.Instance.FromDict(inst_s)
493
    return backend.RunRenameInstance(inst, old_name, debug)
494

    
495
  @staticmethod
496
  def perspective_instance_shutdown(params):
497
    """Shutdown an instance.
498

    
499
    """
500
    instance = objects.Instance.FromDict(params[0])
501
    timeout = params[1]
502
    return backend.InstanceShutdown(instance, timeout)
503

    
504
  @staticmethod
505
  def perspective_instance_start(params):
506
    """Start an instance.
507

    
508
    """
509
    instance = objects.Instance.FromDict(params[0])
510
    return backend.StartInstance(instance)
511

    
512
  @staticmethod
513
  def perspective_migration_info(params):
514
    """Gather information about an instance to be migrated.
515

    
516
    """
517
    instance = objects.Instance.FromDict(params[0])
518
    return backend.MigrationInfo(instance)
519

    
520
  @staticmethod
521
  def perspective_accept_instance(params):
522
    """Prepare the node to accept an instance.
523

    
524
    """
525
    instance, info, target = params
526
    instance = objects.Instance.FromDict(instance)
527
    return backend.AcceptInstance(instance, info, target)
528

    
529
  @staticmethod
530
  def perspective_finalize_migration(params):
531
    """Finalize the instance migration.
532

    
533
    """
534
    instance, info, success = params
535
    instance = objects.Instance.FromDict(instance)
536
    return backend.FinalizeMigration(instance, info, success)
537

    
538
  @staticmethod
539
  def perspective_instance_migrate(params):
540
    """Migrates an instance.
541

    
542
    """
543
    instance, target, live = params
544
    instance = objects.Instance.FromDict(instance)
545
    return backend.MigrateInstance(instance, target, live)
546

    
547
  @staticmethod
548
  def perspective_instance_reboot(params):
549
    """Reboot an instance.
550

    
551
    """
552
    instance = objects.Instance.FromDict(params[0])
553
    reboot_type = params[1]
554
    shutdown_timeout = params[2]
555
    return backend.InstanceReboot(instance, reboot_type, shutdown_timeout)
556

    
557
  @staticmethod
558
  def perspective_instance_info(params):
559
    """Query instance information.
560

    
561
    """
562
    return backend.GetInstanceInfo(params[0], params[1])
563

    
564
  @staticmethod
565
  def perspective_instance_migratable(params):
566
    """Query whether the specified instance can be migrated.
567

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

    
572
  @staticmethod
573
  def perspective_all_instances_info(params):
574
    """Query information about all instances.
575

    
576
    """
577
    return backend.GetAllInstancesInfo(params[0])
578

    
579
  @staticmethod
580
  def perspective_instance_list(params):
581
    """Query the list of running instances.
582

    
583
    """
584
    return backend.GetInstanceList(params[0])
585

    
586
  # node --------------------------
587

    
588
  @staticmethod
589
  def perspective_node_tcp_ping(params):
590
    """Do a TcpPing on the remote node.
591

    
592
    """
593
    return netutils.TcpPing(params[1], params[2], timeout=params[3],
594
                            live_port_needed=params[4], source=params[0])
595

    
596
  @staticmethod
597
  def perspective_node_has_ip_address(params):
598
    """Checks if a node has the given ip address.
599

    
600
    """
601
    return netutils.IPAddress.Own(params[0])
602

    
603
  @staticmethod
604
  def perspective_node_info(params):
605
    """Query node information.
606

    
607
    """
608
    vgname, hypervisor_type = params
609
    return backend.GetNodeInfo(vgname, hypervisor_type)
610

    
611
  @staticmethod
612
  def perspective_etc_hosts_modify(params):
613
    """Modify a node entry in /etc/hosts.
614

    
615
    """
616
    backend.EtcHostsModify(params[0], params[1], params[2])
617

    
618
    return True
619

    
620
  @staticmethod
621
  def perspective_node_verify(params):
622
    """Run a verify sequence on this node.
623

    
624
    """
625
    return backend.VerifyNode(params[0], params[1])
626

    
627
  @staticmethod
628
  def perspective_node_start_master(params):
629
    """Promote this node to master status.
630

    
631
    """
632
    return backend.StartMaster(params[0], params[1])
633

    
634
  @staticmethod
635
  def perspective_node_stop_master(params):
636
    """Demote this node from master status.
637

    
638
    """
639
    return backend.StopMaster(params[0])
640

    
641
  @staticmethod
642
  def perspective_node_leave_cluster(params):
643
    """Cleanup after leaving a cluster.
644

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

    
648
  @staticmethod
649
  def perspective_node_volumes(params):
650
    """Query the list of all logical volume groups.
651

    
652
    """
653
    return backend.NodeVolumes()
654

    
655
  @staticmethod
656
  def perspective_node_demote_from_mc(params):
657
    """Demote a node from the master candidate role.
658

    
659
    """
660
    return backend.DemoteFromMC()
661

    
662

    
663
  @staticmethod
664
  def perspective_node_powercycle(params):
665
    """Tries to powercycle the nod.
666

    
667
    """
668
    hypervisor_type = params[0]
669
    return backend.PowercycleNode(hypervisor_type)
670

    
671

    
672
  # cluster --------------------------
673

    
674
  @staticmethod
675
  def perspective_version(params):
676
    """Query version information.
677

    
678
    """
679
    return constants.PROTOCOL_VERSION
680

    
681
  @staticmethod
682
  def perspective_upload_file(params):
683
    """Upload a file.
684

    
685
    Note that the backend implementation imposes strict rules on which
686
    files are accepted.
687

    
688
    """
689
    return backend.UploadFile(*params)
690

    
691
  @staticmethod
692
  def perspective_master_info(params):
693
    """Query master information.
694

    
695
    """
696
    return backend.GetMasterInfo()
697

    
698
  @staticmethod
699
  def perspective_write_ssconf_files(params):
700
    """Write ssconf files.
701

    
702
    """
703
    (values,) = params
704
    return backend.WriteSsconfFiles(values)
705

    
706
  # os -----------------------
707

    
708
  @staticmethod
709
  def perspective_os_diagnose(params):
710
    """Query detailed information about existing OSes.
711

    
712
    """
713
    return backend.DiagnoseOS()
714

    
715
  @staticmethod
716
  def perspective_os_get(params):
717
    """Query information about a given OS.
718

    
719
    """
720
    name = params[0]
721
    os_obj = backend.OSFromDisk(name)
722
    return os_obj.ToDict()
723

    
724
  @staticmethod
725
  def perspective_os_validate(params):
726
    """Run a given OS' validation routine.
727

    
728
    """
729
    required, name, checks, params = params
730
    return backend.ValidateOS(required, name, checks, params)
731

    
732
  # hooks -----------------------
733

    
734
  @staticmethod
735
  def perspective_hooks_runner(params):
736
    """Run hook scripts.
737

    
738
    """
739
    hpath, phase, env = params
740
    hr = backend.HooksRunner()
741
    return hr.RunHooks(hpath, phase, env)
742

    
743
  # iallocator -----------------
744

    
745
  @staticmethod
746
  def perspective_iallocator_runner(params):
747
    """Run an iallocator script.
748

    
749
    """
750
    name, idata = params
751
    iar = backend.IAllocatorRunner()
752
    return iar.Run(name, idata)
753

    
754
  # test -----------------------
755

    
756
  @staticmethod
757
  def perspective_test_delay(params):
758
    """Run test delay.
759

    
760
    """
761
    duration = params[0]
762
    status, rval = utils.TestDelay(duration)
763
    if not status:
764
      raise backend.RPCFail(rval)
765
    return rval
766

    
767
  # file storage ---------------
768

    
769
  @staticmethod
770
  def perspective_file_storage_dir_create(params):
771
    """Create the file storage directory.
772

    
773
    """
774
    file_storage_dir = params[0]
775
    return backend.CreateFileStorageDir(file_storage_dir)
776

    
777
  @staticmethod
778
  def perspective_file_storage_dir_remove(params):
779
    """Remove the file storage directory.
780

    
781
    """
782
    file_storage_dir = params[0]
783
    return backend.RemoveFileStorageDir(file_storage_dir)
784

    
785
  @staticmethod
786
  def perspective_file_storage_dir_rename(params):
787
    """Rename the file storage directory.
788

    
789
    """
790
    old_file_storage_dir = params[0]
791
    new_file_storage_dir = params[1]
792
    return backend.RenameFileStorageDir(old_file_storage_dir,
793
                                        new_file_storage_dir)
794

    
795
  # jobs ------------------------
796

    
797
  @staticmethod
798
  @_RequireJobQueueLock
799
  def perspective_jobqueue_update(params):
800
    """Update job queue.
801

    
802
    """
803
    (file_name, content) = params
804
    return backend.JobQueueUpdate(file_name, content)
805

    
806
  @staticmethod
807
  @_RequireJobQueueLock
808
  def perspective_jobqueue_purge(params):
809
    """Purge job queue.
810

    
811
    """
812
    return backend.JobQueuePurge()
813

    
814
  @staticmethod
815
  @_RequireJobQueueLock
816
  def perspective_jobqueue_rename(params):
817
    """Rename a job queue file.
818

    
819
    """
820
    # TODO: What if a file fails to rename?
821
    return [backend.JobQueueRename(old, new) for old, new in params]
822

    
823
  # hypervisor ---------------
824

    
825
  @staticmethod
826
  def perspective_hypervisor_validate_params(params):
827
    """Validate the hypervisor parameters.
828

    
829
    """
830
    (hvname, hvparams) = params
831
    return backend.ValidateHVParams(hvname, hvparams)
832

    
833
  # Crypto
834

    
835
  @staticmethod
836
  def perspective_x509_cert_create(params):
837
    """Creates a new X509 certificate for SSL/TLS.
838

    
839
    """
840
    (validity, ) = params
841
    return backend.CreateX509Certificate(validity)
842

    
843
  @staticmethod
844
  def perspective_x509_cert_remove(params):
845
    """Removes a X509 certificate.
846

    
847
    """
848
    (name, ) = params
849
    return backend.RemoveX509Certificate(name)
850

    
851
  # Import and export
852

    
853
  @staticmethod
854
  def perspective_import_start(params):
855
    """Starts an import daemon.
856

    
857
    """
858
    (opts_s, instance, dest, dest_args) = params
859

    
860
    opts = objects.ImportExportOptions.FromDict(opts_s)
861

    
862
    return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
863
                                           None, None,
864
                                           objects.Instance.FromDict(instance),
865
                                           dest,
866
                                           _DecodeImportExportIO(dest,
867
                                                                 dest_args))
868

    
869
  @staticmethod
870
  def perspective_export_start(params):
871
    """Starts an export daemon.
872

    
873
    """
874
    (opts_s, host, port, instance, source, source_args) = params
875

    
876
    opts = objects.ImportExportOptions.FromDict(opts_s)
877

    
878
    return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
879
                                           host, port,
880
                                           objects.Instance.FromDict(instance),
881
                                           source,
882
                                           _DecodeImportExportIO(source,
883
                                                                 source_args))
884

    
885
  @staticmethod
886
  def perspective_impexp_status(params):
887
    """Retrieves the status of an import or export daemon.
888

    
889
    """
890
    return backend.GetImportExportStatus(params[0])
891

    
892
  @staticmethod
893
  def perspective_impexp_abort(params):
894
    """Aborts an import or export.
895

    
896
    """
897
    return backend.AbortImportExport(params[0])
898

    
899
  @staticmethod
900
  def perspective_impexp_cleanup(params):
901
    """Cleans up after an import or export.
902

    
903
    """
904
    return backend.CleanupImportExport(params[0])
905

    
906

    
907
def CheckNoded(_, args):
908
  """Initial checks whether to run or exit with a failure.
909

    
910
  """
911
  if args: # noded doesn't take any arguments
912
    print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
913
                          sys.argv[0])
914
    sys.exit(constants.EXIT_FAILURE)
915

    
916

    
917
def PrepNoded(options, _):
918
  """Preparation node daemon function, executed with the PID file held.
919

    
920
  """
921
  if options.mlock:
922
    request_executor_class = MlockallRequestExecutor
923
    try:
924
      utils.Mlockall()
925
    except errors.NoCtypesError:
926
      logging.warning("Cannot set memory lock, ctypes module not found")
927
      request_executor_class = http.server.HttpServerRequestExecutor
928
  else:
929
    request_executor_class = http.server.HttpServerRequestExecutor
930

    
931
  # Read SSL certificate
932
  if options.ssl:
933
    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
934
                                    ssl_cert_path=options.ssl_cert)
935
  else:
936
    ssl_params = None
937

    
938
  err = _PrepareQueueLock()
939
  if err is not None:
940
    # this might be some kind of file-system/permission error; while
941
    # this breaks the job queue functionality, we shouldn't prevent
942
    # startup of the whole node daemon because of this
943
    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
944

    
945
  mainloop = daemon.Mainloop()
946
  server = NodeHttpServer(mainloop, options.bind_address, options.port,
947
                          ssl_params=ssl_params, ssl_verify_peer=True,
948
                          request_executor_class=request_executor_class)
949
  server.Start()
950
  return (mainloop, server)
951

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

    
955
  """
956
  (mainloop, server) = prep_data
957
  try:
958
    mainloop.Run()
959
  finally:
960
    server.Stop()
961

    
962

    
963
def main():
964
  """Main function for the node daemon.
965

    
966
  """
967
  parser = OptionParser(description="Ganeti node daemon",
968
                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]",
969
                        version="%%prog (ganeti) %s" %
970
                        constants.RELEASE_VERSION)
971
  parser.add_option("--no-mlock", dest="mlock",
972
                    help="Do not mlock the node memory in ram",
973
                    default=True, action="store_false")
974

    
975
  daemon.GenericMain(constants.NODED, parser, CheckNoded, PrepNoded, ExecNoded,
976
                     default_ssl_cert=constants.NODED_CERT_FILE,
977
                     default_ssl_key=constants.NODED_CERT_FILE,
978
                     console_logging=True)
979

    
980

    
981
if __name__ == '__main__':
982
  main()