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 <jenkins@review.openstack.org>
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
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 = (
"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)
]
# 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
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:
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)
'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',
# 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
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:
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.
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:
import random
import signal
import sys
-import threading
import time
try:
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
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:
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:
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)
# 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)
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()
# 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)
os._exit(status)
- LOG.info(_('Started child %d'), pid)
+ LOG.info(_LI('Started child %d'), pid)
wrap.children.add(pid)
self.children[pid] = wrap
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)
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)
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)
# 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()
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
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()
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)
# 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()
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)
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
--- /dev/null
+# 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)
# 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
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).
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
except Exception as ex:
LOG.exception(ex)
+ def stop_timers(self):
for x in self.timers:
try:
x.stop()
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:
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
#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)
Code Review / openstack-build / cinder-build.git / commitdiff