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 <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
This commit is contained in:
Jay S. Bryant
parent
4d024af583
commit
f342c1631b
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
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."))
|
||||
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in New Issue