Statistics
| Branch: | Tag: | Revision:

root / daemons / ganeti-noded @ e8d61457

History | View | Annotate | Download (25.7 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_node_verify(params):
613
    """Run a verify sequence on this node.
614

    
615
    """
616
    return backend.VerifyNode(params[0], params[1])
617

    
618
  @staticmethod
619
  def perspective_node_start_master(params):
620
    """Promote this node to master status.
621

    
622
    """
623
    return backend.StartMaster(params[0], params[1])
624

    
625
  @staticmethod
626
  def perspective_node_stop_master(params):
627
    """Demote this node from master status.
628

    
629
    """
630
    return backend.StopMaster(params[0])
631

    
632
  @staticmethod
633
  def perspective_node_leave_cluster(params):
634
    """Cleanup after leaving a cluster.
635

    
636
    """
637
    return backend.LeaveCluster(params[0])
638

    
639
  @staticmethod
640
  def perspective_node_volumes(params):
641
    """Query the list of all logical volume groups.
642

    
643
    """
644
    return backend.NodeVolumes()
645

    
646
  @staticmethod
647
  def perspective_node_demote_from_mc(params):
648
    """Demote a node from the master candidate role.
649

    
650
    """
651
    return backend.DemoteFromMC()
652

    
653

    
654
  @staticmethod
655
  def perspective_node_powercycle(params):
656
    """Tries to powercycle the nod.
657

    
658
    """
659
    hypervisor_type = params[0]
660
    return backend.PowercycleNode(hypervisor_type)
661

    
662

    
663
  # cluster --------------------------
664

    
665
  @staticmethod
666
  def perspective_version(params):
667
    """Query version information.
668

    
669
    """
670
    return constants.PROTOCOL_VERSION
671

    
672
  @staticmethod
673
  def perspective_upload_file(params):
674
    """Upload a file.
675

    
676
    Note that the backend implementation imposes strict rules on which
677
    files are accepted.
678

    
679
    """
680
    return backend.UploadFile(*params)
681

    
682
  @staticmethod
683
  def perspective_master_info(params):
684
    """Query master information.
685

    
686
    """
687
    return backend.GetMasterInfo()
688

    
689
  @staticmethod
690
  def perspective_write_ssconf_files(params):
691
    """Write ssconf files.
692

    
693
    """
694
    (values,) = params
695
    return backend.WriteSsconfFiles(values)
696

    
697
  # os -----------------------
698

    
699
  @staticmethod
700
  def perspective_os_diagnose(params):
701
    """Query detailed information about existing OSes.
702

    
703
    """
704
    return backend.DiagnoseOS()
705

    
706
  @staticmethod
707
  def perspective_os_get(params):
708
    """Query information about a given OS.
709

    
710
    """
711
    name = params[0]
712
    os_obj = backend.OSFromDisk(name)
713
    return os_obj.ToDict()
714

    
715
  @staticmethod
716
  def perspective_os_validate(params):
717
    """Run a given OS' validation routine.
718

    
719
    """
720
    required, name, checks, params = params
721
    return backend.ValidateOS(required, name, checks, params)
722

    
723
  # hooks -----------------------
724

    
725
  @staticmethod
726
  def perspective_hooks_runner(params):
727
    """Run hook scripts.
728

    
729
    """
730
    hpath, phase, env = params
731
    hr = backend.HooksRunner()
732
    return hr.RunHooks(hpath, phase, env)
733

    
734
  # iallocator -----------------
735

    
736
  @staticmethod
737
  def perspective_iallocator_runner(params):
738
    """Run an iallocator script.
739

    
740
    """
741
    name, idata = params
742
    iar = backend.IAllocatorRunner()
743
    return iar.Run(name, idata)
744

    
745
  # test -----------------------
746

    
747
  @staticmethod
748
  def perspective_test_delay(params):
749
    """Run test delay.
750

    
751
    """
752
    duration = params[0]
753
    status, rval = utils.TestDelay(duration)
754
    if not status:
755
      raise backend.RPCFail(rval)
756
    return rval
757

    
758
  # file storage ---------------
759

    
760
  @staticmethod
761
  def perspective_file_storage_dir_create(params):
762
    """Create the file storage directory.
763

    
764
    """
765
    file_storage_dir = params[0]
766
    return backend.CreateFileStorageDir(file_storage_dir)
767

    
768
  @staticmethod
769
  def perspective_file_storage_dir_remove(params):
770
    """Remove the file storage directory.
771

    
772
    """
773
    file_storage_dir = params[0]
774
    return backend.RemoveFileStorageDir(file_storage_dir)
775

    
776
  @staticmethod
777
  def perspective_file_storage_dir_rename(params):
778
    """Rename the file storage directory.
779

    
780
    """
781
    old_file_storage_dir = params[0]
782
    new_file_storage_dir = params[1]
783
    return backend.RenameFileStorageDir(old_file_storage_dir,
784
                                        new_file_storage_dir)
785

    
786
  # jobs ------------------------
787

    
788
  @staticmethod
789
  @_RequireJobQueueLock
790
  def perspective_jobqueue_update(params):
791
    """Update job queue.
792

    
793
    """
794
    (file_name, content) = params
795
    return backend.JobQueueUpdate(file_name, content)
796

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

    
802
    """
803
    return backend.JobQueuePurge()
804

    
805
  @staticmethod
806
  @_RequireJobQueueLock
807
  def perspective_jobqueue_rename(params):
808
    """Rename a job queue file.
809

    
810
    """
811
    # TODO: What if a file fails to rename?
812
    return [backend.JobQueueRename(old, new) for old, new in params]
813

    
814
  # hypervisor ---------------
815

    
816
  @staticmethod
817
  def perspective_hypervisor_validate_params(params):
818
    """Validate the hypervisor parameters.
819

    
820
    """
821
    (hvname, hvparams) = params
822
    return backend.ValidateHVParams(hvname, hvparams)
823

    
824
  # Crypto
825

    
826
  @staticmethod
827
  def perspective_x509_cert_create(params):
828
    """Creates a new X509 certificate for SSL/TLS.
829

    
830
    """
831
    (validity, ) = params
832
    return backend.CreateX509Certificate(validity)
833

    
834
  @staticmethod
835
  def perspective_x509_cert_remove(params):
836
    """Removes a X509 certificate.
837

    
838
    """
839
    (name, ) = params
840
    return backend.RemoveX509Certificate(name)
841

    
842
  # Import and export
843

    
844
  @staticmethod
845
  def perspective_import_start(params):
846
    """Starts an import daemon.
847

    
848
    """
849
    (opts_s, instance, dest, dest_args) = params
850

    
851
    opts = objects.ImportExportOptions.FromDict(opts_s)
852

    
853
    return backend.StartImportExportDaemon(constants.IEM_IMPORT, opts,
854
                                           None, None,
855
                                           objects.Instance.FromDict(instance),
856
                                           dest,
857
                                           _DecodeImportExportIO(dest,
858
                                                                 dest_args))
859

    
860
  @staticmethod
861
  def perspective_export_start(params):
862
    """Starts an export daemon.
863

    
864
    """
865
    (opts_s, host, port, instance, source, source_args) = params
866

    
867
    opts = objects.ImportExportOptions.FromDict(opts_s)
868

    
869
    return backend.StartImportExportDaemon(constants.IEM_EXPORT, opts,
870
                                           host, port,
871
                                           objects.Instance.FromDict(instance),
872
                                           source,
873
                                           _DecodeImportExportIO(source,
874
                                                                 source_args))
875

    
876
  @staticmethod
877
  def perspective_impexp_status(params):
878
    """Retrieves the status of an import or export daemon.
879

    
880
    """
881
    return backend.GetImportExportStatus(params[0])
882

    
883
  @staticmethod
884
  def perspective_impexp_abort(params):
885
    """Aborts an import or export.
886

    
887
    """
888
    return backend.AbortImportExport(params[0])
889

    
890
  @staticmethod
891
  def perspective_impexp_cleanup(params):
892
    """Cleans up after an import or export.
893

    
894
    """
895
    return backend.CleanupImportExport(params[0])
896

    
897

    
898
def CheckNoded(_, args):
899
  """Initial checks whether to run or exit with a failure.
900

    
901
  """
902
  if args: # noded doesn't take any arguments
903
    print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
904
                          sys.argv[0])
905
    sys.exit(constants.EXIT_FAILURE)
906

    
907

    
908
def ExecNoded(options, _):
909
  """Main node daemon function, executed with the PID file held.
910

    
911
  """
912
  if options.mlock:
913
    request_executor_class = MlockallRequestExecutor
914
    try:
915
      utils.Mlockall()
916
    except errors.NoCtypesError:
917
      logging.warning("Cannot set memory lock, ctypes module not found")
918
      request_executor_class = http.server.HttpServerRequestExecutor
919
  else:
920
    request_executor_class = http.server.HttpServerRequestExecutor
921

    
922
  # Read SSL certificate
923
  if options.ssl:
924
    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
925
                                    ssl_cert_path=options.ssl_cert)
926
  else:
927
    ssl_params = None
928

    
929
  err = _PrepareQueueLock()
930
  if err is not None:
931
    # this might be some kind of file-system/permission error; while
932
    # this breaks the job queue functionality, we shouldn't prevent
933
    # startup of the whole node daemon because of this
934
    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
935

    
936
  mainloop = daemon.Mainloop()
937
  server = NodeHttpServer(mainloop, options.bind_address, options.port,
938
                          ssl_params=ssl_params, ssl_verify_peer=True,
939
                          request_executor_class=request_executor_class)
940
  server.Start()
941
  try:
942
    mainloop.Run()
943
  finally:
944
    server.Stop()
945

    
946

    
947
def main():
948
  """Main function for the node daemon.
949

    
950
  """
951
  parser = OptionParser(description="Ganeti node daemon",
952
                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]",
953
                        version="%%prog (ganeti) %s" %
954
                        constants.RELEASE_VERSION)
955
  parser.add_option("--no-mlock", dest="mlock",
956
                    help="Do not mlock the node memory in ram",
957
                    default=True, action="store_false")
958

    
959
  dirs = [(val, constants.RUN_DIRS_MODE) for val in constants.SUB_RUN_DIRS]
960
  dirs.append((constants.LOG_OS_DIR, 0750))
961
  dirs.append((constants.LOCK_DIR, 1777))
962
  dirs.append((constants.CRYPTO_KEYS_DIR, constants.CRYPTO_KEYS_DIR_MODE))
963
  dirs.append((constants.IMPORT_EXPORT_DIR, constants.IMPORT_EXPORT_DIR_MODE))
964
  daemon.GenericMain(constants.NODED, parser, dirs, CheckNoded, ExecNoded,
965
                     default_ssl_cert=constants.NODED_CERT_FILE,
966
                     default_ssl_key=constants.NODED_CERT_FILE,
967
                     console_logging=True)
968

    
969

    
970
if __name__ == '__main__':
971
  main()