Statistics
| Branch: | Tag: | Revision:

root / daemons / ganeti-noded @ f5118ade

History | View | Annotate | Download (20.8 kB)

1
#!/usr/bin/python
2
#
3

    
4
# Copyright (C) 2006, 2007 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
# functions in this module need to have a given name structure, so:
25
# pylint: disable-msg=C0103
26

    
27
import os
28
import sys
29
import traceback
30
import SocketServer
31
import errno
32
import logging
33
import signal
34

    
35
from optparse import OptionParser
36

    
37
from ganeti import backend
38
from ganeti import constants
39
from ganeti import objects
40
from ganeti import errors
41
from ganeti import jstore
42
from ganeti import daemon
43
from ganeti import http
44
from ganeti import utils
45

    
46
import ganeti.http.server
47

    
48

    
49
queue_lock = None
50

    
51

    
52
def _RequireJobQueueLock(fn):
53
  """Decorator for job queue manipulating functions.
54

    
55
  """
56
  QUEUE_LOCK_TIMEOUT = 10
57

    
58
  def wrapper(*args, **kwargs):
59
    # Locking in exclusive, blocking mode because there could be several
60
    # children running at the same time. Waiting up to 10 seconds.
61
    queue_lock.Exclusive(blocking=True, timeout=QUEUE_LOCK_TIMEOUT)
62
    try:
63
      return fn(*args, **kwargs)
64
    finally:
65
      queue_lock.Unlock()
66

    
67
  return wrapper
68

    
69

    
70
class NodeHttpServer(http.server.HttpServer):
71
  """The server implementation.
72

    
73
  This class holds all methods exposed over the RPC interface.
74

    
75
  """
76
  def __init__(self, *args, **kwargs):
77
    http.server.HttpServer.__init__(self, *args, **kwargs)
78
    self.noded_pid = os.getpid()
79

    
80
  def HandleRequest(self, req):
81
    """Handle a request.
82

    
83
    """
84
    if req.request_method.upper() != http.HTTP_PUT:
85
      raise http.HttpBadRequest()
86

    
87
    path = req.request_path
88
    if path.startswith("/"):
89
      path = path[1:]
90

    
91
    method = getattr(self, "perspective_%s" % path, None)
92
    if method is None:
93
      raise http.HttpNotFound()
94

    
95
    try:
96
      try:
97
        return method(req.request_body)
98
      except:
99
        logging.exception("Error in RPC call")
100
        raise
101
    except errors.QuitGanetiException, err:
102
      # Tell parent to quit
103
      os.kill(self.noded_pid, signal.SIGTERM)
104

    
105
  # the new block devices  --------------------------
106

    
107
  @staticmethod
108
  def perspective_blockdev_create(params):
109
    """Create a block device.
110

    
111
    """
112
    bdev_s, size, owner, on_primary, info = params
113
    bdev = objects.Disk.FromDict(bdev_s)
114
    if bdev is None:
115
      raise ValueError("can't unserialize data!")
116
    return backend.BlockdevCreate(bdev, size, owner, on_primary, info)
117

    
118
  @staticmethod
119
  def perspective_blockdev_remove(params):
120
    """Remove a block device.
121

    
122
    """
123
    bdev_s = params[0]
124
    bdev = objects.Disk.FromDict(bdev_s)
125
    return backend.BlockdevRemove(bdev)
126

    
127
  @staticmethod
128
  def perspective_blockdev_rename(params):
129
    """Remove a block device.
130

    
131
    """
132
    devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params]
133
    return backend.BlockdevRename(devlist)
134

    
135
  @staticmethod
136
  def perspective_blockdev_assemble(params):
137
    """Assemble a block device.
138

    
139
    """
140
    bdev_s, owner, on_primary = params
141
    bdev = objects.Disk.FromDict(bdev_s)
142
    if bdev is None:
143
      raise ValueError("can't unserialize data!")
144
    return backend.BlockdevAssemble(bdev, owner, on_primary)
145

    
146
  @staticmethod
147
  def perspective_blockdev_shutdown(params):
148
    """Shutdown a block device.
149

    
150
    """
151
    bdev_s = params[0]
152
    bdev = objects.Disk.FromDict(bdev_s)
153
    if bdev is None:
154
      raise ValueError("can't unserialize data!")
155
    return backend.BlockdevShutdown(bdev)
156

    
157
  @staticmethod
158
  def perspective_blockdev_addchildren(params):
159
    """Add a child to a mirror device.
160

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

    
164
    """
165
    bdev_s, ndev_s = params
166
    bdev = objects.Disk.FromDict(bdev_s)
167
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
168
    if bdev is None or ndevs.count(None) > 0:
169
      raise ValueError("can't unserialize data!")
170
    return backend.BlockdevAddchildren(bdev, ndevs)
171

    
172
  @staticmethod
173
  def perspective_blockdev_removechildren(params):
174
    """Remove a child from a mirror device.
175

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

    
179
    """
180
    bdev_s, ndev_s = params
181
    bdev = objects.Disk.FromDict(bdev_s)
182
    ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
183
    if bdev is None or ndevs.count(None) > 0:
184
      raise ValueError("can't unserialize data!")
185
    return backend.BlockdevRemovechildren(bdev, ndevs)
186

    
187
  @staticmethod
188
  def perspective_blockdev_getmirrorstatus(params):
189
    """Return the mirror status for a list of disks.
190

    
191
    """
192
    disks = [objects.Disk.FromDict(dsk_s)
193
            for dsk_s in params]
194
    return backend.BlockdevGetmirrorstatus(disks)
195

    
196
  @staticmethod
197
  def perspective_blockdev_find(params):
198
    """Expose the FindBlockDevice functionality for a disk.
199

    
200
    This will try to find but not activate a disk.
201

    
202
    """
203
    disk = objects.Disk.FromDict(params[0])
204
    return backend.BlockdevFind(disk)
205

    
206
  @staticmethod
207
  def perspective_blockdev_snapshot(params):
208
    """Create a snapshot device.
209

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

    
214
    """
215
    cfbd = objects.Disk.FromDict(params[0])
216
    return backend.BlockdevSnapshot(cfbd)
217

    
218
  @staticmethod
219
  def perspective_blockdev_grow(params):
220
    """Grow a stack of devices.
221

    
222
    """
223
    cfbd = objects.Disk.FromDict(params[0])
224
    amount = params[1]
225
    return backend.BlockdevGrow(cfbd, amount)
226

    
227
  @staticmethod
228
  def perspective_blockdev_close(params):
229
    """Closes the given block devices.
230

    
231
    """
232
    disks = [objects.Disk.FromDict(cf) for cf in params[1]]
233
    return backend.BlockdevClose(params[0], disks)
234

    
235
  # blockdev/drbd specific methods ----------
236

    
237
  @staticmethod
238
  def perspective_drbd_disconnect_net(params):
239
    """Disconnects the network connection of drbd disks.
240

    
241
    Note that this is only valid for drbd disks, so the members of the
242
    disk list must all be drbd devices.
243

    
244
    """
245
    nodes_ip, disks = params
246
    disks = [objects.Disk.FromDict(cf) for cf in disks]
247
    return backend.DrbdDisconnectNet(nodes_ip, disks)
248

    
249
  @staticmethod
250
  def perspective_drbd_attach_net(params):
251
    """Attaches the network connection of drbd disks.
252

    
253
    Note that this is only valid for drbd disks, so the members of the
254
    disk list must all be drbd devices.
255

    
256
    """
257
    nodes_ip, disks, instance_name, multimaster = params
258
    disks = [objects.Disk.FromDict(cf) for cf in disks]
259
    return backend.DrbdAttachNet(nodes_ip, disks,
260
                                     instance_name, multimaster)
261

    
262
  @staticmethod
263
  def perspective_drbd_wait_sync(params):
264
    """Wait until DRBD disks are synched.
265

    
266
    Note that this is only valid for drbd disks, so the members of the
267
    disk list must all be drbd devices.
268

    
269
    """
270
    nodes_ip, disks = params
271
    disks = [objects.Disk.FromDict(cf) for cf in disks]
272
    return backend.DrbdWaitSync(nodes_ip, disks)
273

    
274
  # export/import  --------------------------
275

    
276
  @staticmethod
277
  def perspective_snapshot_export(params):
278
    """Export a given snapshot.
279

    
280
    """
281
    disk = objects.Disk.FromDict(params[0])
282
    dest_node = params[1]
283
    instance = objects.Instance.FromDict(params[2])
284
    cluster_name = params[3]
285
    dev_idx = params[4]
286
    return backend.ExportSnapshot(disk, dest_node, instance,
287
                                  cluster_name, dev_idx)
288

    
289
  @staticmethod
290
  def perspective_finalize_export(params):
291
    """Expose the finalize export functionality.
292

    
293
    """
294
    instance = objects.Instance.FromDict(params[0])
295
    snap_disks = [objects.Disk.FromDict(str_data)
296
                  for str_data in params[1]]
297
    return backend.FinalizeExport(instance, snap_disks)
298

    
299
  @staticmethod
300
  def perspective_export_info(params):
301
    """Query information about an existing export on this node.
302

    
303
    The given path may not contain an export, in which case we return
304
    None.
305

    
306
    """
307
    path = params[0]
308
    einfo = backend.ExportInfo(path)
309
    if einfo is None:
310
      return einfo
311
    return einfo.Dumps()
312

    
313
  @staticmethod
314
  def perspective_export_list(params):
315
    """List the available exports on this node.
316

    
317
    Note that as opposed to export_info, which may query data about an
318
    export in any path, this only queries the standard Ganeti path
319
    (constants.EXPORT_DIR).
320

    
321
    """
322
    return backend.ListExports()
323

    
324
  @staticmethod
325
  def perspective_export_remove(params):
326
    """Remove an export.
327

    
328
    """
329
    export = params[0]
330
    return backend.RemoveExport(export)
331

    
332
  # volume  --------------------------
333

    
334
  @staticmethod
335
  def perspective_volume_list(params):
336
    """Query the list of logical volumes in a given volume group.
337

    
338
    """
339
    vgname = params[0]
340
    return backend.GetVolumeList(vgname)
341

    
342
  @staticmethod
343
  def perspective_vg_list(params):
344
    """Query the list of volume groups.
345

    
346
    """
347
    return backend.ListVolumeGroups()
348

    
349
  # bridge  --------------------------
350

    
351
  @staticmethod
352
  def perspective_bridges_exist(params):
353
    """Check if all bridges given exist on this node.
354

    
355
    """
356
    bridges_list = params[0]
357
    return backend.BridgesExist(bridges_list)
358

    
359
  # instance  --------------------------
360

    
361
  @staticmethod
362
  def perspective_instance_os_add(params):
363
    """Install an OS on a given instance.
364

    
365
    """
366
    inst_s = params[0]
367
    inst = objects.Instance.FromDict(inst_s)
368
    reinstall = params[1]
369
    return backend.InstanceOsAdd(inst, reinstall)
370

    
371
  @staticmethod
372
  def perspective_instance_run_rename(params):
373
    """Runs the OS rename script for an instance.
374

    
375
    """
376
    inst_s, old_name = params
377
    inst = objects.Instance.FromDict(inst_s)
378
    return backend.RunRenameInstance(inst, old_name)
379

    
380
  @staticmethod
381
  def perspective_instance_os_import(params):
382
    """Run the import function of an OS onto a given instance.
383

    
384
    """
385
    inst_s, src_node, src_images, cluster_name = params
386
    inst = objects.Instance.FromDict(inst_s)
387
    return backend.ImportOSIntoInstance(inst, src_node, src_images,
388
                                        cluster_name)
389

    
390
  @staticmethod
391
  def perspective_instance_shutdown(params):
392
    """Shutdown an instance.
393

    
394
    """
395
    instance = objects.Instance.FromDict(params[0])
396
    return backend.InstanceShutdown(instance)
397

    
398
  @staticmethod
399
  def perspective_instance_start(params):
400
    """Start an instance.
401

    
402
    """
403
    instance = objects.Instance.FromDict(params[0])
404
    return backend.StartInstance(instance)
405

    
406
  @staticmethod
407
  def perspective_migration_info(params):
408
    """Gather information about an instance to be migrated.
409

    
410
    """
411
    instance = objects.Instance.FromDict(params[0])
412
    return backend.MigrationInfo(instance)
413

    
414
  @staticmethod
415
  def perspective_accept_instance(params):
416
    """Prepare the node to accept an instance.
417

    
418
    """
419
    instance, info, target = params
420
    instance = objects.Instance.FromDict(instance)
421
    return backend.AcceptInstance(instance, info, target)
422

    
423
  @staticmethod
424
  def perspective_finalize_migration(params):
425
    """Finalize the instance migration.
426

    
427
    """
428
    instance, info, success = params
429
    instance = objects.Instance.FromDict(instance)
430
    return backend.FinalizeMigration(instance, info, success)
431

    
432
  @staticmethod
433
  def perspective_instance_migrate(params):
434
    """Migrates an instance.
435

    
436
    """
437
    instance, target, live = params
438
    instance = objects.Instance.FromDict(instance)
439
    return backend.MigrateInstance(instance, target, live)
440

    
441
  @staticmethod
442
  def perspective_instance_reboot(params):
443
    """Reboot an instance.
444

    
445
    """
446
    instance = objects.Instance.FromDict(params[0])
447
    reboot_type = params[1]
448
    return backend.InstanceReboot(instance, reboot_type)
449

    
450
  @staticmethod
451
  def perspective_instance_info(params):
452
    """Query instance information.
453

    
454
    """
455
    return backend.GetInstanceInfo(params[0], params[1])
456

    
457
  @staticmethod
458
  def perspective_instance_migratable(params):
459
    """Query whether the specified instance can be migrated.
460

    
461
    """
462
    instance = objects.Instance.FromDict(params[0])
463
    return backend.GetInstanceMigratable(instance)
464

    
465
  @staticmethod
466
  def perspective_all_instances_info(params):
467
    """Query information about all instances.
468

    
469
    """
470
    return backend.GetAllInstancesInfo(params[0])
471

    
472
  @staticmethod
473
  def perspective_instance_list(params):
474
    """Query the list of running instances.
475

    
476
    """
477
    return backend.GetInstanceList(params[0])
478

    
479
  # node --------------------------
480

    
481
  @staticmethod
482
  def perspective_node_tcp_ping(params):
483
    """Do a TcpPing on the remote node.
484

    
485
    """
486
    return utils.TcpPing(params[1], params[2], timeout=params[3],
487
                         live_port_needed=params[4], source=params[0])
488

    
489
  @staticmethod
490
  def perspective_node_has_ip_address(params):
491
    """Checks if a node has the given ip address.
492

    
493
    """
494
    return utils.OwnIpAddress(params[0])
495

    
496
  @staticmethod
497
  def perspective_node_info(params):
498
    """Query node information.
499

    
500
    """
501
    vgname, hypervisor_type = params
502
    return backend.GetNodeInfo(vgname, hypervisor_type)
503

    
504
  @staticmethod
505
  def perspective_node_add(params):
506
    """Complete the registration of this node in the cluster.
507

    
508
    """
509
    return backend.AddNode(params[0], params[1], params[2],
510
                           params[3], params[4], params[5])
511

    
512
  @staticmethod
513
  def perspective_node_verify(params):
514
    """Run a verify sequence on this node.
515

    
516
    """
517
    return backend.VerifyNode(params[0], params[1])
518

    
519
  @staticmethod
520
  def perspective_node_start_master(params):
521
    """Promote this node to master status.
522

    
523
    """
524
    return backend.StartMaster(params[0])
525

    
526
  @staticmethod
527
  def perspective_node_stop_master(params):
528
    """Demote this node from master status.
529

    
530
    """
531
    return backend.StopMaster(params[0])
532

    
533
  @staticmethod
534
  def perspective_node_leave_cluster(params):
535
    """Cleanup after leaving a cluster.
536

    
537
    """
538
    return backend.LeaveCluster()
539

    
540
  @staticmethod
541
  def perspective_node_volumes(params):
542
    """Query the list of all logical volume groups.
543

    
544
    """
545
    return backend.NodeVolumes()
546

    
547
  @staticmethod
548
  def perspective_node_demote_from_mc(params):
549
    """Demote a node from the master candidate role.
550

    
551
    """
552
    return backend.DemoteFromMC()
553

    
554

    
555
  @staticmethod
556
  def perspective_node_powercycle(params):
557
    """Tries to powercycle the nod.
558

    
559
    """
560
    hypervisor_type = params[0]
561
    return backend.PowercycleNode(hypervisor_type)
562

    
563

    
564
  # cluster --------------------------
565

    
566
  @staticmethod
567
  def perspective_version(params):
568
    """Query version information.
569

    
570
    """
571
    return constants.PROTOCOL_VERSION
572

    
573
  @staticmethod
574
  def perspective_upload_file(params):
575
    """Upload a file.
576

    
577
    Note that the backend implementation imposes strict rules on which
578
    files are accepted.
579

    
580
    """
581
    return backend.UploadFile(*params)
582

    
583
  @staticmethod
584
  def perspective_master_info(params):
585
    """Query master information.
586

    
587
    """
588
    return backend.GetMasterInfo()
589

    
590
  @staticmethod
591
  def perspective_write_ssconf_files(params):
592
    """Write ssconf files.
593

    
594
    """
595
    (values,) = params
596
    return backend.WriteSsconfFiles(values)
597

    
598
  # os -----------------------
599

    
600
  @staticmethod
601
  def perspective_os_diagnose(params):
602
    """Query detailed information about existing OSes.
603

    
604
    """
605
    return [os_obj.ToDict() for os_obj in backend.DiagnoseOS()]
606

    
607
  @staticmethod
608
  def perspective_os_get(params):
609
    """Query information about a given OS.
610

    
611
    """
612
    name = params[0]
613
    try:
614
      os_obj = backend.OSFromDisk(name)
615
    except errors.InvalidOS, err:
616
      os_obj = objects.OS.FromInvalidOS(err)
617
    return os_obj.ToDict()
618

    
619
  # hooks -----------------------
620

    
621
  @staticmethod
622
  def perspective_hooks_runner(params):
623
    """Run hook scripts.
624

    
625
    """
626
    hpath, phase, env = params
627
    hr = backend.HooksRunner()
628
    return hr.RunHooks(hpath, phase, env)
629

    
630
  # iallocator -----------------
631

    
632
  @staticmethod
633
  def perspective_iallocator_runner(params):
634
    """Run an iallocator script.
635

    
636
    """
637
    name, idata = params
638
    iar = backend.IAllocatorRunner()
639
    return iar.Run(name, idata)
640

    
641
  # test -----------------------
642

    
643
  @staticmethod
644
  def perspective_test_delay(params):
645
    """Run test delay.
646

    
647
    """
648
    duration = params[0]
649
    return utils.TestDelay(duration)
650

    
651
  # file storage ---------------
652

    
653
  @staticmethod
654
  def perspective_file_storage_dir_create(params):
655
    """Create the file storage directory.
656

    
657
    """
658
    file_storage_dir = params[0]
659
    return backend.CreateFileStorageDir(file_storage_dir)
660

    
661
  @staticmethod
662
  def perspective_file_storage_dir_remove(params):
663
    """Remove the file storage directory.
664

    
665
    """
666
    file_storage_dir = params[0]
667
    return backend.RemoveFileStorageDir(file_storage_dir)
668

    
669
  @staticmethod
670
  def perspective_file_storage_dir_rename(params):
671
    """Rename the file storage directory.
672

    
673
    """
674
    old_file_storage_dir = params[0]
675
    new_file_storage_dir = params[1]
676
    return backend.RenameFileStorageDir(old_file_storage_dir,
677
                                        new_file_storage_dir)
678

    
679
  # jobs ------------------------
680

    
681
  @staticmethod
682
  @_RequireJobQueueLock
683
  def perspective_jobqueue_update(params):
684
    """Update job queue.
685

    
686
    """
687
    (file_name, content) = params
688
    return backend.JobQueueUpdate(file_name, content)
689

    
690
  @staticmethod
691
  @_RequireJobQueueLock
692
  def perspective_jobqueue_purge(params):
693
    """Purge job queue.
694

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

    
698
  @staticmethod
699
  @_RequireJobQueueLock
700
  def perspective_jobqueue_rename(params):
701
    """Rename a job queue file.
702

    
703
    """
704
    # TODO: What if a file fails to rename?
705
    return [backend.JobQueueRename(old, new) for old, new in params]
706

    
707
  @staticmethod
708
  def perspective_jobqueue_set_drain(params):
709
    """Set/unset the queue drain flag.
710

    
711
    """
712
    drain_flag = params[0]
713
    return backend.JobQueueSetDrainFlag(drain_flag)
714

    
715

    
716
  # hypervisor ---------------
717

    
718
  @staticmethod
719
  def perspective_hypervisor_validate_params(params):
720
    """Validate the hypervisor parameters.
721

    
722
    """
723
    (hvname, hvparams) = params
724
    return backend.ValidateHVParams(hvname, hvparams)
725

    
726

    
727
def ParseOptions():
728
  """Parse the command line options.
729

    
730
  @return: (options, args) as from OptionParser.parse_args()
731

    
732
  """
733
  parser = OptionParser(description="Ganeti node daemon",
734
                        usage="%prog [-f] [-d] [-b ADDRESS]",
735
                        version="%%prog (ganeti) %s" %
736
                        constants.RELEASE_VERSION)
737

    
738
  parser.add_option("-f", "--foreground", dest="fork",
739
                    help="Don't detach from the current terminal",
740
                    default=True, action="store_false")
741
  parser.add_option("-d", "--debug", dest="debug",
742
                    help="Enable some debug messages",
743
                    default=False, action="store_true")
744
  parser.add_option("-b", "--bind", dest="bind_address",
745
                    help="Bind address",
746
                    default="", metavar="ADDRESS")
747

    
748
  options, args = parser.parse_args()
749
  return options, args
750

    
751

    
752
def main():
753
  """Main function for the node daemon.
754

    
755
  """
756
  global queue_lock
757

    
758
  options, args = ParseOptions()
759
  utils.debug = options.debug
760

    
761
  if options.fork:
762
    utils.CloseFDs()
763

    
764
  for fname in (constants.SSL_CERT_FILE,):
765
    if not os.path.isfile(fname):
766
      print "config %s not there, will not run." % fname
767
      sys.exit(5)
768

    
769
  try:
770
    port = utils.GetNodeDaemonPort()
771
  except errors.ConfigurationError, err:
772
    print "Cluster configuration incomplete: '%s'" % str(err)
773
    sys.exit(5)
774

    
775
  dirs = [(val, constants.RUN_DIRS_MODE) for val in constants.SUB_RUN_DIRS]
776
  dirs.append((constants.LOG_OS_DIR, 0750))
777
  dirs.append((constants.LOCK_DIR, 1777))
778
  utils.EnsureDirs(dirs)
779

    
780
  # become a daemon
781
  if options.fork:
782
    utils.Daemonize(logfile=constants.LOG_NODESERVER)
783

    
784
  utils.WritePidFile(constants.NODED_PID)
785
  try:
786
    utils.SetupLogging(logfile=constants.LOG_NODESERVER, debug=options.debug,
787
                       stderr_logging=not options.fork)
788
    logging.info("ganeti node daemon startup")
789

    
790
    # Read SSL certificate
791
    ssl_params = http.HttpSslParams(ssl_key_path=constants.SSL_CERT_FILE,
792
                                    ssl_cert_path=constants.SSL_CERT_FILE)
793

    
794
    # Prepare job queue
795
    queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
796

    
797
    mainloop = daemon.Mainloop()
798
    server = NodeHttpServer(mainloop, options.bind_address, port,
799
                            ssl_params=ssl_params, ssl_verify_peer=True)
800
    server.Start()
801
    try:
802
      mainloop.Run()
803
    finally:
804
      server.Stop()
805
  finally:
806
    utils.RemovePidFile(constants.NODED_PID)
807

    
808

    
809
if __name__ == '__main__':
810
  main()