From: Jay S. Bryant Date: Wed, 10 Sep 2014 15:18:22 +0000 (-0500) Subject: Sync service.py and its dependencies to Cinder X-Git-Url: https://review.fuel-infra.org/gitweb?a=commitdiff_plain;h=f342c1631b619f1239dedbe3aad93c4b474c4f55;p=openstack-build%2Fcinder-build.git Sync service.py and its dependencies to Cinder Bug 1362455 made note that our sevice.py was out of sync from what is currently in oslo-incubator. We have been syncing things piece meal. It would be good to get this code synced up. ---------------------------------------------- Head of oslo-incubator: commit 820752a10d8277464261661ca211f2c3e7282ce0 Merge: 8ebe497 422e94b Author: Jenkins Date: Wed Sep 10 14:14:28 2014 +0000 Merge "Switch to using pbr's autodoc capability" ----------------------------------------------- This commit pulls in the following changes listed by file, ordered from newest to oldest: cinder/openstack/common/eventlet_backdoor.py ------------------------------------------------ 90ae24bf Remove redundant default=None for config options fcf517d7 Update oslo log messages with translation domains cinder/openstack/common/jsonutils.py ------------------------------------------------ d9060f69 Check for namedtuple_as_object support before using it ad248f66 Specify namedtuple_as_object=False when using simplejson cinder/openstack/common/log.py ------------------------------------------------ 037dee00 Set stevedore log level to WARN by default cinder/openstack/common/loopingcall.py ------------------------------------------------ e3773930 Changes calcuation of variable delay ab5d5f1c Use timestamp in loopingcall bc48099a Log the function name of looping call fb4e863c Remove deprecated LoopingCall fcf517d7 Update oslo log messages with translation domains cinder/openstack/common/service.py ------------------------------------------------ 5f12bba4 Implement stop method in ProcessLauncher 8c00b361 Merge "Remove duplicated "caught" message" f718e404 Remove duplicated "caught" message 356ecd4d Move notification point to a better place e2634a75 Add missing _LI for LOG.info in service module 0150ad75 Merge "Reap child processes gracefully if greenlet thread gets killed" 53e1214c notify calling process we are ready to serve a6fec438 Reap child processes gracefully if greenlet thread gets killed fcf517d7 Update oslo log messages with translation domains 1e700782 Revert "service: replace eventlet event by threading" 0644073e Simplify launch method 6b4d255e Merge "service: replace eventlet event by threading" a4f145e4 Merge "Allow configurable ProcessLauncher liveness check" cinder/openstack/common/threadgroup.py ------------------------------------------------ 5a1a0166 Make stop_timers() method public fdc88831 Add graceful stop function to ThreadGroup.stop 5f8ace05 Merge "threadgroup: use threading rather than greenthread" 2d06d6ca Simple typo correction 4d18b57a threadgroup: use threading rather than greenthread cinder/openstack/common/systemd.py ------------------------------------------------- 17c4e21e Fix docstring indentation in systemd 667d1bac Fixed spelling error - occured to occurred 53e1214c notify calling process we are ready to serve Change-Id: I163f0b80c32c3b8485d9c007fa6e4474494ba31f Closes-bug: 1362455 --- diff --git a/cinder/openstack/common/eventlet_backdoor.py b/cinder/openstack/common/eventlet_backdoor.py index f1526f6b0..ec9505f7d 100644 --- a/cinder/openstack/common/eventlet_backdoor.py +++ b/cinder/openstack/common/eventlet_backdoor.py @@ -29,7 +29,7 @@ import eventlet.backdoor import greenlet from oslo.config import cfg -from cinder.openstack.common.gettextutils import _ +from cinder.openstack.common.gettextutils import _LI from cinder.openstack.common import log as logging help_for_backdoor_port = ( @@ -41,7 +41,6 @@ help_for_backdoor_port = ( "chosen port is displayed in the service's log file.") eventlet_backdoor_opts = [ cfg.StrOpt('backdoor_port', - default=None, help="Enable eventlet backdoor. %s" % help_for_backdoor_port) ] @@ -137,8 +136,10 @@ def initialize_if_enabled(): # In the case of backdoor port being zero, a port number is assigned by # listen(). In any case, pull the port number out here. port = sock.getsockname()[1] - LOG.info(_('Eventlet backdoor listening on %(port)s for process %(pid)d') % - {'port': port, 'pid': os.getpid()}) + LOG.info( + _LI('Eventlet backdoor listening on %(port)s for process %(pid)d') % + {'port': port, 'pid': os.getpid()} + ) eventlet.spawn_n(eventlet.backdoor.backdoor_server, sock, locals=backdoor_locals) return port diff --git a/cinder/openstack/common/jsonutils.py b/cinder/openstack/common/jsonutils.py index 744d63435..1ee1fff33 100644 --- a/cinder/openstack/common/jsonutils.py +++ b/cinder/openstack/common/jsonutils.py @@ -38,11 +38,19 @@ import inspect import itertools import sys +is_simplejson = False if sys.version_info < (2, 7): # On Python <= 2.6, json module is not C boosted, so try to use # simplejson module if available try: import simplejson as json + # NOTE(mriedem): Make sure we have a new enough version of simplejson + # to support the namedobject_as_tuple argument. This can be removed + # in the Kilo release when python 2.6 support is dropped. + if 'namedtuple_as_object' in inspect.getargspec(json.dumps).args: + is_simplejson = True + else: + import json except ImportError: import json else: @@ -165,10 +173,14 @@ def to_primitive(value, convert_instances=False, convert_datetime=True, def dumps(value, default=to_primitive, **kwargs): + if is_simplejson: + kwargs['namedtuple_as_object'] = False return json.dumps(value, default=default, **kwargs) def dump(obj, fp, *args, **kwargs): + if is_simplejson: + kwargs['namedtuple_as_object'] = False return json.dump(obj, fp, *args, **kwargs) diff --git a/cinder/openstack/common/log.py b/cinder/openstack/common/log.py index 9b7b7fde3..f325925a9 100644 --- a/cinder/openstack/common/log.py +++ b/cinder/openstack/common/log.py @@ -128,7 +128,8 @@ DEFAULT_LOG_LEVELS = ['amqp=WARN', 'amqplib=WARN', 'boto=WARN', 'oslo.messaging=INFO', 'iso8601=WARN', 'requests.packages.urllib3.connectionpool=WARN', 'urllib3.connectionpool=WARN', 'websocket=WARN', - "keystonemiddleware=WARN", "routes.middleware=WARN"] + "keystonemiddleware=WARN", "routes.middleware=WARN", + "stevedore=WARN"] log_opts = [ cfg.StrOpt('logging_context_format_string', diff --git a/cinder/openstack/common/loopingcall.py b/cinder/openstack/common/loopingcall.py index eabb605ff..ec1200153 100644 --- a/cinder/openstack/common/loopingcall.py +++ b/cinder/openstack/common/loopingcall.py @@ -16,31 +16,36 @@ # under the License. import sys +import time from eventlet import event from eventlet import greenthread -from cinder.openstack.common.gettextutils import _ +from cinder.openstack.common.gettextutils import _LE, _LW from cinder.openstack.common import log as logging -from cinder.openstack.common import timeutils LOG = logging.getLogger(__name__) +# NOTE(zyluo): This lambda function was declared to avoid mocking collisions +# with time.time() called in the standard logging module +# during unittests. +_ts = lambda: time.time() + class LoopingCallDone(Exception): - """Exception to break out and stop a LoopingCall. + """Exception to break out and stop a LoopingCallBase. - The poll-function passed to LoopingCall can raise this exception to + The poll-function passed to LoopingCallBase can raise this exception to break out of the loop normally. This is somewhat analogous to StopIteration. An optional return-value can be included as the argument to the exception; - this return-value will be returned by LoopingCall.wait() + this return-value will be returned by LoopingCallBase.wait() """ def __init__(self, retvalue=True): - """:param retvalue: Value that LoopingCall.wait() should return.""" + """:param retvalue: Value that LoopingCallBase.wait() should return.""" self.retvalue = retvalue @@ -72,21 +77,22 @@ class FixedIntervalLoopingCall(LoopingCallBase): try: while self._running: - start = timeutils.utcnow() + start = _ts() self.f(*self.args, **self.kw) - end = timeutils.utcnow() + end = _ts() if not self._running: break - delay = interval - timeutils.delta_seconds(start, end) - if delay <= 0: - LOG.warn(_('task run outlasted interval by %s sec') % - -delay) - greenthread.sleep(delay if delay > 0 else 0) + delay = end - start - interval + if delay > 0: + LOG.warn(_LW('task %(func_name)s run outlasted ' + 'interval by %(delay).2f sec'), + {'func_name': repr(self.f), 'delay': delay}) + greenthread.sleep(-delay if delay < 0 else 0) except LoopingCallDone as e: self.stop() done.send(e.retvalue) except Exception: - LOG.exception(_('in fixed duration looping call')) + LOG.exception(_LE('in fixed duration looping call')) done.send_exception(*sys.exc_info()) return else: @@ -98,11 +104,6 @@ class FixedIntervalLoopingCall(LoopingCallBase): return self.done -# TODO(mikal): this class name is deprecated in Havana and should be removed -# in the I release -LoopingCall = FixedIntervalLoopingCall - - class DynamicLoopingCall(LoopingCallBase): """A looping call which sleeps until the next known event. @@ -126,14 +127,15 @@ class DynamicLoopingCall(LoopingCallBase): if periodic_interval_max is not None: idle = min(idle, periodic_interval_max) - LOG.debug(_('Dynamic looping call sleeping for %.02f ' - 'seconds'), idle) + LOG.debug('Dynamic looping call %(func_name)s sleeping ' + 'for %(idle).02f seconds', + {'func_name': repr(self.f), 'idle': idle}) greenthread.sleep(idle) except LoopingCallDone as e: self.stop() done.send(e.retvalue) except Exception: - LOG.exception(_('in dynamic looping call')) + LOG.exception(_LE('in dynamic looping call')) done.send_exception(*sys.exc_info()) return else: diff --git a/cinder/openstack/common/service.py b/cinder/openstack/common/service.py index 7632625d4..a63d38b3e 100644 --- a/cinder/openstack/common/service.py +++ b/cinder/openstack/common/service.py @@ -23,7 +23,6 @@ import os import random import signal import sys -import threading import time try: @@ -35,12 +34,14 @@ except ImportError: UnsupportedOperation = None import eventlet +from eventlet import event from oslo.config import cfg from cinder.openstack.common import eventlet_backdoor -from cinder.openstack.common.gettextutils import _ +from cinder.openstack.common.gettextutils import _LE, _LI, _LW from cinder.openstack.common import importutils from cinder.openstack.common import log as logging +from cinder.openstack.common import systemd from cinder.openstack.common import threadgroup @@ -163,7 +164,7 @@ class ServiceLauncher(Launcher): status = None signo = 0 - LOG.debug(_('Full set of CONF:')) + LOG.debug('Full set of CONF:') CONF.log_opt_values(LOG, std_logging.DEBUG) try: @@ -172,7 +173,7 @@ class ServiceLauncher(Launcher): super(ServiceLauncher, self).wait() except SignalExit as exc: signame = _signo_to_signame(exc.signo) - LOG.info(_('Caught %s, exiting'), signame) + LOG.info(_LI('Caught %s, exiting'), signame) status = exc.code signo = exc.signo except SystemExit as exc: @@ -184,11 +185,12 @@ class ServiceLauncher(Launcher): rpc.cleanup() except Exception: # We're shutting down, so it doesn't matter at this point. - LOG.exception(_('Exception during rpc cleanup.')) + LOG.exception(_LE('Exception during rpc cleanup.')) return status, signo def wait(self, ready_callback=None): + systemd.notify_once() while True: self.handle_signal() status, signo = self._wait_for_exit_or_signal(ready_callback) @@ -235,7 +237,7 @@ class ProcessLauncher(object): # dies unexpectedly self.readpipe.read() - LOG.info(_('Parent process has died unexpectedly, exiting')) + LOG.info(_LI('Parent process has died unexpectedly, exiting')) sys.exit(1) @@ -266,13 +268,13 @@ class ProcessLauncher(object): launcher.wait() except SignalExit as exc: signame = _signo_to_signame(exc.signo) - LOG.info(_('Caught %s, exiting'), signame) + LOG.info(_LI('Child caught %s, exiting'), signame) status = exc.code signo = exc.signo except SystemExit as exc: status = exc.code except BaseException: - LOG.exception(_('Unhandled exception')) + LOG.exception(_LE('Unhandled exception')) status = 2 finally: launcher.stop() @@ -305,7 +307,7 @@ class ProcessLauncher(object): # start up quickly but ensure we don't fork off children that # die instantly too quickly. if time.time() - wrap.forktimes[0] < wrap.workers: - LOG.info(_('Forking too fast, sleeping')) + LOG.info(_LI('Forking too fast, sleeping')) time.sleep(1) wrap.forktimes.pop(0) @@ -324,7 +326,7 @@ class ProcessLauncher(object): os._exit(status) - LOG.info(_('Started child %d'), pid) + LOG.info(_LI('Started child %d'), pid) wrap.children.add(pid) self.children[pid] = wrap @@ -334,7 +336,7 @@ class ProcessLauncher(object): def launch_service(self, service, workers=1): wrap = ServiceWrapper(service, workers) - LOG.info(_('Starting %d workers'), wrap.workers) + LOG.info(_LI('Starting %d workers'), wrap.workers) while self.running and len(wrap.children) < wrap.workers: self._start_child(wrap) @@ -351,15 +353,15 @@ class ProcessLauncher(object): if os.WIFSIGNALED(status): sig = os.WTERMSIG(status) - LOG.info(_('Child %(pid)d killed by signal %(sig)d'), + LOG.info(_LI('Child %(pid)d killed by signal %(sig)d'), dict(pid=pid, sig=sig)) else: code = os.WEXITSTATUS(status) - LOG.info(_('Child %(pid)s exited with status %(code)d'), + LOG.info(_LI('Child %(pid)s exited with status %(code)d'), dict(pid=pid, code=code)) if pid not in self.children: - LOG.warning(_('pid %d not in child list'), pid) + LOG.warning(_LW('pid %d not in child list'), pid) return None wrap = self.children.pop(pid) @@ -381,23 +383,35 @@ class ProcessLauncher(object): def wait(self): """Loop waiting on children to die and respawning as necessary.""" - LOG.debug(_('Full set of CONF:')) + systemd.notify_once() + LOG.debug('Full set of CONF:') CONF.log_opt_values(LOG, std_logging.DEBUG) - while True: - self.handle_signal() - self._respawn_children() - if self.sigcaught: + try: + while True: + self.handle_signal() + self._respawn_children() + # No signal means that stop was called. Don't clean up here. + if not self.sigcaught: + return + signame = _signo_to_signame(self.sigcaught) - LOG.info(_('Caught %s, stopping children'), signame) - if not _is_sighup_and_daemon(self.sigcaught): - break + LOG.info(_LI('Caught %s, stopping children'), signame) + if not _is_sighup_and_daemon(self.sigcaught): + break + + for pid in self.children: + os.kill(pid, signal.SIGHUP) + self.running = True + self.sigcaught = None + except eventlet.greenlet.GreenletExit: + LOG.info(_LI("Wait called after thread killed. Cleaning up.")) - for pid in self.children: - os.kill(pid, signal.SIGHUP) - self.running = True - self.sigcaught = None + self.stop() + def stop(self): + """Terminate child processes and wait on each.""" + self.running = False for pid in self.children: try: os.kill(pid, signal.SIGTERM) @@ -407,7 +421,7 @@ class ProcessLauncher(object): # Wait for children to die if self.children: - LOG.info(_('Waiting on %d children to exit'), len(self.children)) + LOG.info(_LI('Waiting on %d children to exit'), len(self.children)) while self.children: self._wait_child() @@ -419,10 +433,11 @@ class Service(object): self.tg = threadgroup.ThreadGroup(threads) # signal that the service is done shutting itself down: - self._done = threading.Event() + self._done = event.Event() def reset(self): - self._done = threading.Event() + # NOTE(Fengqian): docs for Event.reset() recommend against using it + self._done = event.Event() def start(self): pass @@ -431,7 +446,8 @@ class Service(object): self.tg.stop() self.tg.wait() # Signal that service cleanup is done: - self._done.set() + if not self._done.ready(): + self._done.send() def wait(self): self._done.wait() @@ -442,7 +458,7 @@ class Services(object): def __init__(self): self.services = [] self.tg = threadgroup.ThreadGroup() - self.done = threading.Event() + self.done = event.Event() def add(self, service): self.services.append(service) @@ -456,7 +472,8 @@ class Services(object): # Each service has performed cleanup, now signal that the run_service # wrapper threads can now die: - self.done.set() + if not self.done.ready(): + self.done.send() # reap threads: self.tg.stop() @@ -466,7 +483,7 @@ class Services(object): def restart(self): self.stop() - self.done = threading.Event() + self.done = event.Event() for restart_service in self.services: restart_service.reset() self.tg.add_thread(self.run_service, restart_service, self.done) @@ -484,11 +501,12 @@ class Services(object): done.wait() -def launch(service, workers=None): - if workers: - launcher = ProcessLauncher() - launcher.launch_service(service, workers=workers) - else: +def launch(service, workers=1): + if workers is None or workers == 1: launcher = ServiceLauncher() launcher.launch_service(service) + else: + launcher = ProcessLauncher() + launcher.launch_service(service, workers=workers) + return launcher diff --git a/cinder/openstack/common/systemd.py b/cinder/openstack/common/systemd.py new file mode 100644 index 000000000..d60befd72 --- /dev/null +++ b/cinder/openstack/common/systemd.py @@ -0,0 +1,106 @@ +# Copyright 2012-2014 Red Hat, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +""" +Helper module for systemd service readiness notification. +""" + +import os +import socket +import sys + +from cinder.openstack.common import log as logging + + +LOG = logging.getLogger(__name__) + + +def _abstractify(socket_name): + if socket_name.startswith('@'): + # abstract namespace socket + socket_name = '\0%s' % socket_name[1:] + return socket_name + + +def _sd_notify(unset_env, msg): + notify_socket = os.getenv('NOTIFY_SOCKET') + if notify_socket: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + try: + sock.connect(_abstractify(notify_socket)) + sock.sendall(msg) + if unset_env: + del os.environ['NOTIFY_SOCKET'] + except EnvironmentError: + LOG.debug("Systemd notification failed", exc_info=True) + finally: + sock.close() + + +def notify(): + """Send notification to Systemd that service is ready. + + For details see + http://www.freedesktop.org/software/systemd/man/sd_notify.html + """ + _sd_notify(False, 'READY=1') + + +def notify_once(): + """Send notification once to Systemd that service is ready. + + Systemd sets NOTIFY_SOCKET environment variable with the name of the + socket listening for notifications from services. + This method removes the NOTIFY_SOCKET environment variable to ensure + notification is sent only once. + """ + _sd_notify(True, 'READY=1') + + +def onready(notify_socket, timeout): + """Wait for systemd style notification on the socket. + + :param notify_socket: local socket address + :type notify_socket: string + :param timeout: socket timeout + :type timeout: float + :returns: 0 service ready + 1 service not ready + 2 timeout occurred + """ + sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + sock.settimeout(timeout) + sock.bind(_abstractify(notify_socket)) + try: + msg = sock.recv(512) + except socket.timeout: + return 2 + finally: + sock.close() + if 'READY=1' in msg: + return 0 + else: + return 1 + + +if __name__ == '__main__': + # simple CLI for testing + if len(sys.argv) == 1: + notify() + elif len(sys.argv) >= 2: + timeout = float(sys.argv[1]) + notify_socket = os.getenv('NOTIFY_SOCKET') + if notify_socket: + retval = onready(notify_socket, timeout) + sys.exit(retval) diff --git a/cinder/openstack/common/threadgroup.py b/cinder/openstack/common/threadgroup.py index edeaf3070..12acf5df9 100644 --- a/cinder/openstack/common/threadgroup.py +++ b/cinder/openstack/common/threadgroup.py @@ -11,10 +11,10 @@ # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. +import threading import eventlet from eventlet import greenpool -from eventlet import greenthread from cinder.openstack.common import log as logging from cinder.openstack.common import loopingcall @@ -51,7 +51,7 @@ class Thread(object): class ThreadGroup(object): - """The point of the ThreadGroup classis to: + """The point of the ThreadGroup class is to: * keep track of timers and greenthreads (making it easier to stop them when need be). @@ -85,8 +85,8 @@ class ThreadGroup(object): def thread_done(self, thread): self.threads.remove(thread) - def stop(self): - current = greenthread.getcurrent() + def _stop_threads(self): + current = threading.current_thread() # Iterate over a copy of self.threads so thread_done doesn't # modify the list while we're iterating @@ -99,6 +99,7 @@ class ThreadGroup(object): except Exception as ex: LOG.exception(ex) + def stop_timers(self): for x in self.timers: try: x.stop() @@ -106,6 +107,23 @@ class ThreadGroup(object): LOG.exception(ex) self.timers = [] + def stop(self, graceful=False): + """stop function has the option of graceful=True/False. + + * In case of graceful=True, wait for all threads to be finished. + Never kill threads. + * In case of graceful=False, kill threads immediately. + """ + self.stop_timers() + if graceful: + # In case of graceful=True, wait for all threads to be + # finished, never kill threads + self.wait() + else: + # In case of graceful=False(Default), kill threads + # immediately + self._stop_threads() + def wait(self): for x in self.timers: try: @@ -114,7 +132,7 @@ class ThreadGroup(object): pass except Exception as ex: LOG.exception(ex) - current = greenthread.getcurrent() + current = threading.current_thread() # Iterate over a copy of self.threads so thread_done doesn't # modify the list while we're iterating diff --git a/etc/cinder/cinder.conf.sample b/etc/cinder/cinder.conf.sample index a78853d52..7101f4bc9 100644 --- a/etc/cinder/cinder.conf.sample +++ b/etc/cinder/cinder.conf.sample @@ -798,7 +798,7 @@ #logging_exception_prefix=%(asctime)s.%(msecs)03d %(process)d TRACE %(name)s %(instance)s # List of logger=LEVEL pairs. (list value) -#default_log_levels=amqp=WARN,amqplib=WARN,boto=WARN,qpid=WARN,sqlalchemy=WARN,suds=INFO,oslo.messaging=INFO,iso8601=WARN,requests.packages.urllib3.connectionpool=WARN,urllib3.connectionpool=WARN,websocket=WARN,keystonemiddleware=WARN,routes.middleware=WARN +#default_log_levels=amqp=WARN,amqplib=WARN,boto=WARN,qpid=WARN,sqlalchemy=WARN,suds=INFO,oslo.messaging=INFO,iso8601=WARN,requests.packages.urllib3.connectionpool=WARN,urllib3.connectionpool=WARN,websocket=WARN,keystonemiddleware=WARN,routes.middleware=WARN,stevedore=WARN # Enables or disables publication of error events. (boolean # value)