Statistics
| Branch: | Tag: | Revision:

root / lib / daemon.py @ b604d0c8

History | View | Annotate | Download (23.5 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2010, 2011 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
"""Module with helper classes and functions for daemons"""
23

    
24

    
25
import asyncore
26
import asynchat
27
import collections
28
import os
29
import signal
30
import logging
31
import sched
32
import time
33
import socket
34
import select
35
import sys
36

    
37
from ganeti import utils
38
from ganeti import constants
39
from ganeti import errors
40
from ganeti import netutils
41
from ganeti import ssconf
42
from ganeti import runtime
43
from ganeti import compat
44

    
45

    
46
class SchedulerBreakout(Exception):
47
  """Exception used to get out of the scheduler loop
48

49
  """
50

    
51

    
52
def AsyncoreDelayFunction(timeout):
53
  """Asyncore-compatible scheduler delay function.
54

55
  This is a delay function for sched that, rather than actually sleeping,
56
  executes asyncore events happening in the meantime.
57

58
  After an event has occurred, rather than returning, it raises a
59
  SchedulerBreakout exception, which will force the current scheduler.run()
60
  invocation to terminate, so that we can also check for signals. The main loop
61
  will then call the scheduler run again, which will allow it to actually
62
  process any due events.
63

64
  This is needed because scheduler.run() doesn't support a count=..., as
65
  asyncore loop, and the scheduler module documents throwing exceptions from
66
  inside the delay function as an allowed usage model.
67

68
  """
69
  asyncore.loop(timeout=timeout, count=1, use_poll=True)
70
  raise SchedulerBreakout()
71

    
72

    
73
class AsyncoreScheduler(sched.scheduler):
74
  """Event scheduler integrated with asyncore
75

76
  """
77
  def __init__(self, timefunc):
78
    sched.scheduler.__init__(self, timefunc, AsyncoreDelayFunction)
79

    
80

    
81
class GanetiBaseAsyncoreDispatcher(asyncore.dispatcher):
82
  """Base Ganeti Asyncore Dispacher
83

84
  """
85
  # this method is overriding an asyncore.dispatcher method
86
  def handle_error(self):
87
    """Log an error in handling any request, and proceed.
88

89
    """
90
    logging.exception("Error while handling asyncore request")
91

    
92
  # this method is overriding an asyncore.dispatcher method
93
  def writable(self):
94
    """Most of the time we don't want to check for writability.
95

96
    """
97
    return False
98

    
99

    
100
class AsyncStreamServer(GanetiBaseAsyncoreDispatcher):
101
  """A stream server to use with asyncore.
102

103
  Each request is accepted, and then dispatched to a separate asyncore
104
  dispatcher to handle.
105

106
  """
107

    
108
  _REQUEST_QUEUE_SIZE = 5
109

    
110
  def __init__(self, family, address):
111
    """Constructor for AsyncUnixStreamSocket
112

113
    @type family: integer
114
    @param family: socket family (one of socket.AF_*)
115
    @type address: address family dependent
116
    @param address: address to bind the socket to
117

118
    """
119
    GanetiBaseAsyncoreDispatcher.__init__(self)
120
    self.family = family
121
    self.create_socket(self.family, socket.SOCK_STREAM)
122
    self.set_reuse_addr()
123
    self.bind(address)
124
    self.listen(self._REQUEST_QUEUE_SIZE)
125

    
126
  # this method is overriding an asyncore.dispatcher method
127
  def handle_accept(self):
128
    """Accept a new client connection.
129

130
    Creates a new instance of the handler class, which will use asyncore to
131
    serve the client.
132

133
    """
134
    accept_result = utils.IgnoreSignals(self.accept)
135
    if accept_result is not None:
136
      connected_socket, client_address = accept_result
137
      if self.family == socket.AF_UNIX:
138
        # override the client address, as for unix sockets nothing meaningful
139
        # is passed in from accept anyway
140
        client_address = netutils.GetSocketCredentials(connected_socket)
141
      logging.info("Accepted connection from %s",
142
                   netutils.FormatAddress(client_address, family=self.family))
143
      self.handle_connection(connected_socket, client_address)
144

    
145
  def handle_connection(self, connected_socket, client_address):
146
    """Handle an already accepted connection.
147

148
    """
149
    raise NotImplementedError
150

    
151

    
152
class AsyncTerminatedMessageStream(asynchat.async_chat):
153
  """A terminator separated message stream asyncore module.
154

155
  Handles a stream connection receiving messages terminated by a defined
156
  separator. For each complete message handle_message is called.
157

158
  """
159
  def __init__(self, connected_socket, peer_address, terminator, family,
160
               unhandled_limit):
161
    """AsyncTerminatedMessageStream constructor.
162

163
    @type connected_socket: socket.socket
164
    @param connected_socket: connected stream socket to receive messages from
165
    @param peer_address: family-specific peer address
166
    @type terminator: string
167
    @param terminator: terminator separating messages in the stream
168
    @type family: integer
169
    @param family: socket family
170
    @type unhandled_limit: integer or None
171
    @param unhandled_limit: maximum unanswered messages
172

173
    """
174
    # python 2.4/2.5 uses conn=... while 2.6 has sock=... we have to cheat by
175
    # using a positional argument rather than a keyword one.
176
    asynchat.async_chat.__init__(self, connected_socket)
177
    self.connected_socket = connected_socket
178
    # on python 2.4 there is no "family" attribute for the socket class
179
    # FIXME: when we move to python 2.5 or above remove the family parameter
180
    #self.family = self.connected_socket.family
181
    self.family = family
182
    self.peer_address = peer_address
183
    self.terminator = terminator
184
    self.unhandled_limit = unhandled_limit
185
    self.set_terminator(terminator)
186
    self.ibuffer = []
187
    self.receive_count = 0
188
    self.send_count = 0
189
    self.oqueue = collections.deque()
190
    self.iqueue = collections.deque()
191

    
192
  # this method is overriding an asynchat.async_chat method
193
  def collect_incoming_data(self, data):
194
    self.ibuffer.append(data)
195

    
196
  def _can_handle_message(self):
197
    return (self.unhandled_limit is None or
198
            (self.receive_count < self.send_count + self.unhandled_limit) and
199
             not self.iqueue)
200

    
201
  # this method is overriding an asynchat.async_chat method
202
  def found_terminator(self):
203
    message = "".join(self.ibuffer)
204
    self.ibuffer = []
205
    message_id = self.receive_count
206
    # We need to increase the receive_count after checking if the message can
207
    # be handled, but before calling handle_message
208
    can_handle = self._can_handle_message()
209
    self.receive_count += 1
210
    if can_handle:
211
      self.handle_message(message, message_id)
212
    else:
213
      self.iqueue.append((message, message_id))
214

    
215
  def handle_message(self, message, message_id):
216
    """Handle a terminated message.
217

218
    @type message: string
219
    @param message: message to handle
220
    @type message_id: integer
221
    @param message_id: stream's message sequence number
222

223
    """
224
    pass
225
    # TODO: move this method to raise NotImplementedError
226
    # raise NotImplementedError
227

    
228
  def send_message(self, message):
229
    """Send a message to the remote peer. This function is thread-safe.
230

231
    @type message: string
232
    @param message: message to send, without the terminator
233

234
    @warning: If calling this function from a thread different than the one
235
    performing the main asyncore loop, remember that you have to wake that one
236
    up.
237

238
    """
239
    # If we just append the message we received to the output queue, this
240
    # function can be safely called by multiple threads at the same time, and
241
    # we don't need locking, since deques are thread safe. handle_write in the
242
    # asyncore thread will handle the next input message if there are any
243
    # enqueued.
244
    self.oqueue.append(message)
245

    
246
  # this method is overriding an asyncore.dispatcher method
247
  def readable(self):
248
    # read from the socket if we can handle the next requests
249
    return self._can_handle_message() and asynchat.async_chat.readable(self)
250

    
251
  # this method is overriding an asyncore.dispatcher method
252
  def writable(self):
253
    # the output queue may become full just after we called writable. This only
254
    # works if we know we'll have something else waking us up from the select,
255
    # in such case, anyway.
256
    return asynchat.async_chat.writable(self) or self.oqueue
257

    
258
  # this method is overriding an asyncore.dispatcher method
259
  def handle_write(self):
260
    if self.oqueue:
261
      # if we have data in the output queue, then send_message was called.
262
      # this means we can process one more message from the input queue, if
263
      # there are any.
264
      data = self.oqueue.popleft()
265
      self.push(data + self.terminator)
266
      self.send_count += 1
267
      if self.iqueue:
268
        self.handle_message(*self.iqueue.popleft())
269
    self.initiate_send()
270

    
271
  def close_log(self):
272
    logging.info("Closing connection from %s",
273
                 netutils.FormatAddress(self.peer_address, family=self.family))
274
    self.close()
275

    
276
  # this method is overriding an asyncore.dispatcher method
277
  def handle_expt(self):
278
    self.close_log()
279

    
280
  # this method is overriding an asyncore.dispatcher method
281
  def handle_error(self):
282
    """Log an error in handling any request, and proceed.
283

284
    """
285
    logging.exception("Error while handling asyncore request")
286
    self.close_log()
287

    
288

    
289
class AsyncUDPSocket(GanetiBaseAsyncoreDispatcher):
290
  """An improved asyncore udp socket.
291

292
  """
293
  def __init__(self, family):
294
    """Constructor for AsyncUDPSocket
295

296
    """
297
    GanetiBaseAsyncoreDispatcher.__init__(self)
298
    self._out_queue = []
299
    self._family = family
300
    self.create_socket(family, socket.SOCK_DGRAM)
301

    
302
  # this method is overriding an asyncore.dispatcher method
303
  def handle_connect(self):
304
    # Python thinks that the first udp message from a source qualifies as a
305
    # "connect" and further ones are part of the same connection. We beg to
306
    # differ and treat all messages equally.
307
    pass
308

    
309
  # this method is overriding an asyncore.dispatcher method
310
  def handle_read(self):
311
    recv_result = utils.IgnoreSignals(self.recvfrom,
312
                                      constants.MAX_UDP_DATA_SIZE)
313
    if recv_result is not None:
314
      payload, address = recv_result
315
      if self._family == socket.AF_INET6:
316
        # we ignore 'flow info' and 'scope id' as we don't need them
317
        ip, port, _, _ = address
318
      else:
319
        ip, port = address
320

    
321
      self.handle_datagram(payload, ip, port)
322

    
323
  def handle_datagram(self, payload, ip, port):
324
    """Handle an already read udp datagram
325

326
    """
327
    raise NotImplementedError
328

    
329
  # this method is overriding an asyncore.dispatcher method
330
  def writable(self):
331
    # We should check whether we can write to the socket only if we have
332
    # something scheduled to be written
333
    return bool(self._out_queue)
334

    
335
  # this method is overriding an asyncore.dispatcher method
336
  def handle_write(self):
337
    if not self._out_queue:
338
      logging.error("handle_write called with empty output queue")
339
      return
340
    (ip, port, payload) = self._out_queue[0]
341
    utils.IgnoreSignals(self.sendto, payload, 0, (ip, port))
342
    self._out_queue.pop(0)
343

    
344
  def enqueue_send(self, ip, port, payload):
345
    """Enqueue a datagram to be sent when possible
346

347
    """
348
    if len(payload) > constants.MAX_UDP_DATA_SIZE:
349
      raise errors.UdpDataSizeError('Packet too big: %s > %s' % (len(payload),
350
                                    constants.MAX_UDP_DATA_SIZE))
351
    self._out_queue.append((ip, port, payload))
352

    
353
  def process_next_packet(self, timeout=0):
354
    """Process the next datagram, waiting for it if necessary.
355

356
    @type timeout: float
357
    @param timeout: how long to wait for data
358
    @rtype: boolean
359
    @return: True if some data has been handled, False otherwise
360

361
    """
362
    result = utils.WaitForFdCondition(self, select.POLLIN, timeout)
363
    if result is not None and result & select.POLLIN:
364
      self.handle_read()
365
      return True
366
    else:
367
      return False
368

    
369

    
370
class AsyncAwaker(GanetiBaseAsyncoreDispatcher):
371
  """A way to notify the asyncore loop that something is going on.
372

373
  If an asyncore daemon is multithreaded when a thread tries to push some data
374
  to a socket, the main loop handling asynchronous requests might be sleeping
375
  waiting on a select(). To avoid this it can create an instance of the
376
  AsyncAwaker, which other threads can use to wake it up.
377

378
  """
379
  def __init__(self, signal_fn=None):
380
    """Constructor for AsyncAwaker
381

382
    @type signal_fn: function
383
    @param signal_fn: function to call when awaken
384

385
    """
386
    GanetiBaseAsyncoreDispatcher.__init__(self)
387
    assert signal_fn == None or callable(signal_fn)
388
    (self.in_socket, self.out_socket) = socket.socketpair(socket.AF_UNIX,
389
                                                          socket.SOCK_STREAM)
390
    self.in_socket.setblocking(0)
391
    self.in_socket.shutdown(socket.SHUT_WR)
392
    self.out_socket.shutdown(socket.SHUT_RD)
393
    self.set_socket(self.in_socket)
394
    self.need_signal = True
395
    self.signal_fn = signal_fn
396
    self.connected = True
397

    
398
  # this method is overriding an asyncore.dispatcher method
399
  def handle_read(self):
400
    utils.IgnoreSignals(self.recv, 4096)
401
    if self.signal_fn:
402
      self.signal_fn()
403
    self.need_signal = True
404

    
405
  # this method is overriding an asyncore.dispatcher method
406
  def close(self):
407
    asyncore.dispatcher.close(self)
408
    self.out_socket.close()
409

    
410
  def signal(self):
411
    """Signal the asyncore main loop.
412

413
    Any data we send here will be ignored, but it will cause the select() call
414
    to return.
415

416
    """
417
    # Yes, there is a race condition here. No, we don't care, at worst we're
418
    # sending more than one wakeup token, which doesn't harm at all.
419
    if self.need_signal:
420
      self.need_signal = False
421
      self.out_socket.send("\0")
422

    
423

    
424
class Mainloop(object):
425
  """Generic mainloop for daemons
426

427
  @ivar scheduler: A sched.scheduler object, which can be used to register
428
    timed events
429

430
  """
431
  def __init__(self):
432
    """Constructs a new Mainloop instance.
433

434
    """
435
    self._signal_wait = []
436
    self.scheduler = AsyncoreScheduler(time.time)
437

    
438
    # Resolve uid/gids used
439
    runtime.GetEnts()
440

    
441
  @utils.SignalHandled([signal.SIGCHLD])
442
  @utils.SignalHandled([signal.SIGTERM])
443
  @utils.SignalHandled([signal.SIGINT])
444
  def Run(self, signal_handlers=None):
445
    """Runs the mainloop.
446

447
    @type signal_handlers: dict
448
    @param signal_handlers: signal->L{utils.SignalHandler} passed by decorator
449

450
    """
451
    assert isinstance(signal_handlers, dict) and \
452
           len(signal_handlers) > 0, \
453
           "Broken SignalHandled decorator"
454
    running = True
455

    
456
    # Start actual main loop
457
    while running:
458
      if not self.scheduler.empty():
459
        try:
460
          self.scheduler.run()
461
        except SchedulerBreakout:
462
          pass
463
      else:
464
        asyncore.loop(count=1, use_poll=True)
465

    
466
      # Check whether a signal was raised
467
      for sig in signal_handlers:
468
        handler = signal_handlers[sig]
469
        if handler.called:
470
          self._CallSignalWaiters(sig)
471
          running = sig not in (signal.SIGTERM, signal.SIGINT)
472
          handler.Clear()
473

    
474
  def _CallSignalWaiters(self, signum):
475
    """Calls all signal waiters for a certain signal.
476

477
    @type signum: int
478
    @param signum: Signal number
479

480
    """
481
    for owner in self._signal_wait:
482
      owner.OnSignal(signum)
483

    
484
  def RegisterSignal(self, owner):
485
    """Registers a receiver for signal notifications
486

487
    The receiver must support a "OnSignal(self, signum)" function.
488

489
    @type owner: instance
490
    @param owner: Receiver
491

492
    """
493
    self._signal_wait.append(owner)
494

    
495

    
496
def _VerifyDaemonUser(daemon_name):
497
  """Verifies the process uid matches the configured uid.
498

499
  This method verifies that a daemon is started as the user it is
500
  intended to be run
501

502
  @param daemon_name: The name of daemon to be started
503
  @return: A tuple with the first item indicating success or not,
504
           the second item current uid and third with expected uid
505

506
  """
507
  getents = runtime.GetEnts()
508
  running_uid = os.getuid()
509
  daemon_uids = {
510
    constants.MASTERD: getents.masterd_uid,
511
    constants.RAPI: getents.rapi_uid,
512
    constants.NODED: getents.noded_uid,
513
    constants.CONFD: getents.confd_uid,
514
    }
515

    
516
  return (daemon_uids[daemon_name] == running_uid, running_uid,
517
          daemon_uids[daemon_name])
518

    
519

    
520
def _BeautifyError(err):
521
  """Try to format an error better.
522

523
  Since we're dealing with daemon startup errors, in many cases this
524
  will be due to socket error and such, so we try to format these cases better.
525

526
  @param err: an exception object
527
  @rtype: string
528
  @return: the formatted error description
529

530
  """
531
  try:
532
    if isinstance(err, socket.error):
533
      return "Socket-related error: %s (errno=%s)" % (err.args[1], err.args[0])
534
    elif isinstance(err, EnvironmentError):
535
      if err.filename is None:
536
        return "%s (errno=%s)" % (err.strerror, err.errno)
537
      else:
538
        return "%s (file %s) (errno=%s)" % (err.strerror, err.filename,
539
                                            err.errno)
540
    else:
541
      return str(err)
542
  except Exception: # pylint: disable-msg=W0703
543
    logging.exception("Error while handling existing error %s", err)
544
    return "%s" % str(err)
545

    
546

    
547
def _HandleSigHup(reopen_cb, signum, frame): # pylint: disable-msg=W0613
548
  """Handler for SIGHUP.
549

550
  @param reopen_cb: Callback function for reopening log files
551

552
  """
553
  assert callable(reopen_cb)
554
  logging.info("Reopening log files after receiving SIGHUP")
555
  reopen_cb()
556

    
557

    
558
def GenericMain(daemon_name, optionparser,
559
                check_fn, prepare_fn, exec_fn,
560
                multithreaded=False, console_logging=False,
561
                default_ssl_cert=None, default_ssl_key=None):
562
  """Shared main function for daemons.
563

564
  @type daemon_name: string
565
  @param daemon_name: daemon name
566
  @type optionparser: optparse.OptionParser
567
  @param optionparser: initialized optionparser with daemon-specific options
568
                       (common -f -d options will be handled by this module)
569
  @type check_fn: function which accepts (options, args)
570
  @param check_fn: function that checks start conditions and exits if they're
571
                   not met
572
  @type prepare_fn: function which accepts (options, args)
573
  @param prepare_fn: function that is run before forking, or None;
574
      it's result will be passed as the third parameter to exec_fn, or
575
      if None was passed in, we will just pass None to exec_fn
576
  @type exec_fn: function which accepts (options, args, prepare_results)
577
  @param exec_fn: function that's executed with the daemon's pid file held, and
578
                  runs the daemon itself.
579
  @type multithreaded: bool
580
  @param multithreaded: Whether the daemon uses threads
581
  @type console_logging: boolean
582
  @param console_logging: if True, the daemon will fall back to the system
583
                          console if logging fails
584
  @type default_ssl_cert: string
585
  @param default_ssl_cert: Default SSL certificate path
586
  @type default_ssl_key: string
587
  @param default_ssl_key: Default SSL key path
588

589
  """
590
  optionparser.add_option("-f", "--foreground", dest="fork",
591
                          help="Don't detach from the current terminal",
592
                          default=True, action="store_false")
593
  optionparser.add_option("-d", "--debug", dest="debug",
594
                          help="Enable some debug messages",
595
                          default=False, action="store_true")
596
  optionparser.add_option("--syslog", dest="syslog",
597
                          help="Enable logging to syslog (except debug"
598
                          " messages); one of 'no', 'yes' or 'only' [%s]" %
599
                          constants.SYSLOG_USAGE,
600
                          default=constants.SYSLOG_USAGE,
601
                          choices=["no", "yes", "only"])
602

    
603
  if daemon_name in constants.DAEMONS_PORTS:
604
    default_bind_address = constants.IP4_ADDRESS_ANY
605
    family = ssconf.SimpleStore().GetPrimaryIPFamily()
606
    # family will default to AF_INET if there is no ssconf file (e.g. when
607
    # upgrading a cluster from 2.2 -> 2.3. This is intended, as Ganeti clusters
608
    # <= 2.2 can not be AF_INET6
609
    if family == netutils.IP6Address.family:
610
      default_bind_address = constants.IP6_ADDRESS_ANY
611

    
612
    default_port = netutils.GetDaemonPort(daemon_name)
613

    
614
    # For networked daemons we allow choosing the port and bind address
615
    optionparser.add_option("-p", "--port", dest="port",
616
                            help="Network port (default: %s)" % default_port,
617
                            default=default_port, type="int")
618
    optionparser.add_option("-b", "--bind", dest="bind_address",
619
                            help=("Bind address (default: '%s')" %
620
                                  default_bind_address),
621
                            default=default_bind_address, metavar="ADDRESS")
622

    
623
  if default_ssl_key is not None and default_ssl_cert is not None:
624
    optionparser.add_option("--no-ssl", dest="ssl",
625
                            help="Do not secure HTTP protocol with SSL",
626
                            default=True, action="store_false")
627
    optionparser.add_option("-K", "--ssl-key", dest="ssl_key",
628
                            help=("SSL key path (default: %s)" %
629
                                  default_ssl_key),
630
                            default=default_ssl_key, type="string",
631
                            metavar="SSL_KEY_PATH")
632
    optionparser.add_option("-C", "--ssl-cert", dest="ssl_cert",
633
                            help=("SSL certificate path (default: %s)" %
634
                                  default_ssl_cert),
635
                            default=default_ssl_cert, type="string",
636
                            metavar="SSL_CERT_PATH")
637

    
638
  # Disable the use of fork(2) if the daemon uses threads
639
  if multithreaded:
640
    utils.DisableFork()
641

    
642
  options, args = optionparser.parse_args()
643

    
644
  if getattr(options, "ssl", False):
645
    ssl_paths = {
646
      "certificate": options.ssl_cert,
647
      "key": options.ssl_key,
648
      }
649

    
650
    for name, path in ssl_paths.iteritems():
651
      if not os.path.isfile(path):
652
        print >> sys.stderr, "SSL %s file '%s' was not found" % (name, path)
653
        sys.exit(constants.EXIT_FAILURE)
654

    
655
    # TODO: By initiating http.HttpSslParams here we would only read the files
656
    # once and have a proper validation (isfile returns False on directories)
657
    # at the same time.
658

    
659
  result, running_uid, expected_uid = _VerifyDaemonUser(daemon_name)
660
  if not result:
661
    msg = ("%s started using wrong user ID (%d), expected %d" %
662
           (daemon_name, running_uid, expected_uid))
663
    print >> sys.stderr, msg
664
    sys.exit(constants.EXIT_FAILURE)
665

    
666
  if check_fn is not None:
667
    check_fn(options, args)
668

    
669
  if options.fork:
670
    utils.CloseFDs()
671
    wpipe = utils.Daemonize(logfile=constants.DAEMONS_LOGFILES[daemon_name])
672
  else:
673
    wpipe = None
674

    
675
  log_reopen_fn = \
676
    utils.SetupLogging(constants.DAEMONS_LOGFILES[daemon_name], daemon_name,
677
                       debug=options.debug,
678
                       stderr_logging=not options.fork,
679
                       multithreaded=multithreaded,
680
                       syslog=options.syslog,
681
                       console_logging=console_logging)
682

    
683
  # Reopen log file(s) on SIGHUP
684
  signal.signal(signal.SIGHUP, compat.partial(_HandleSigHup, log_reopen_fn))
685

    
686
  utils.WritePidFile(utils.DaemonPidFileName(daemon_name))
687
  try:
688
    try:
689
      logging.info("%s daemon startup", daemon_name)
690
      if callable(prepare_fn):
691
        prep_results = prepare_fn(options, args)
692
      else:
693
        prep_results = None
694
    except Exception, err:
695
      utils.WriteErrorToFD(wpipe, _BeautifyError(err))
696
      raise
697

    
698
    if wpipe is not None:
699
      # we're done with the preparation phase, we close the pipe to
700
      # let the parent know it's safe to exit
701
      os.close(wpipe)
702

    
703
    exec_fn(options, args, prep_results)
704
  finally:
705
    utils.RemoveFile(utils.DaemonPidFileName(daemon_name))