Add worker lifecycle DEBUG logging decorator
Introduce ``log_worker_lifecycle`` in ``neutron.common.utils`` to log when worker start, wait, stop, and reset methods are called and finished. Apply it to ``PeriodicWorker``, ``RpcWorker``, ``AllServicesNeutronWorker`` and ``BGPWorker``, so operators can trace child process behavior at DEBUG level. Related-Bug: #2154192 Signed-off-by: Rodolfo Alonso Hernandez <ralonsoh@redhat.com> Change-Id: Iaac0993da9ecf0d1b587bb8c01e3683a0994b7c2
This commit is contained in:
1 parent
b6d260e10c
commit
aa18156e29
6 files changed
+106
-9
No files matched your search
@@ -174,6 +174,36 @@ class exception_logger:
|
||||
return call
|
||||
|
||||
|
||||
def log_worker_lifecycle(description, finished_only=False):
|
||||
"""Log worker lifecycle method calls at DEBUG level.
|
||||
|
||||
:param description: Callable receiving the worker instance and returning
|
||||
the description logged in the message, or the name of an instance
|
||||
attribute holding that value.
|
||||
:param finished_only: If True, only log the method finished message.
|
||||
"""
|
||||
def decorator(func):
|
||||
method_name = func.__name__
|
||||
|
||||
@functools.wraps(func)
|
||||
def wrapper(self, *args, **kwargs):
|
||||
log = logging.getLogger(func.__module__)
|
||||
worker_class = self.__class__.__name__
|
||||
if callable(description):
|
||||
desc = description(self)
|
||||
else:
|
||||
desc = getattr(self, description, None)
|
||||
if not finished_only:
|
||||
log.debug('%s %s called, description: %s',
|
||||
worker_class, method_name, desc)
|
||||
result = func(self, *args, **kwargs)
|
||||
log.debug('%s %s finished, description: %s',
|
||||
worker_class, method_name, desc)
|
||||
return result
|
||||
return wrapper
|
||||
return decorator
|
||||
|
||||
|
||||
def get_other_dvr_serviced_device_owners(host_dvr_for_dhcp=True):
|
||||
"""Return device_owner names for ports that should be serviced by DVR
|
||||
|
||||
|
||||
+14
-7
@@ -38,6 +38,7 @@ import psutil
|
||||
from neutron._i18n import _
|
||||
from neutron.common import config
|
||||
from neutron.common import profiler
|
||||
from neutron.common import utils as common_utils
|
||||
from neutron.conf import service
|
||||
from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import worker as \
|
||||
ovn_worker
|
||||
@@ -64,6 +65,7 @@ class RpcWorker(neutron_worker.NeutronBaseWorker):
|
||||
self._plugins = plugins
|
||||
self._servers = []
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def start(self):
|
||||
super().start(desc=self.desc)
|
||||
for plugin in self._plugins:
|
||||
@@ -74,6 +76,7 @@ class RpcWorker(neutron_worker.NeutronBaseWorker):
|
||||
continue
|
||||
self._servers.extend(servers)
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def wait(self):
|
||||
try:
|
||||
self._wait()
|
||||
@@ -82,24 +85,22 @@ class RpcWorker(neutron_worker.NeutronBaseWorker):
|
||||
raise
|
||||
|
||||
def _wait(self):
|
||||
LOG.debug('calling RpcWorker wait()')
|
||||
for server in self._servers:
|
||||
if isinstance(server, rpc_server.MessageHandlingServer):
|
||||
LOG.debug('calling wait on %s', server)
|
||||
server.wait()
|
||||
else:
|
||||
LOG.debug('NOT calling wait on %s', server)
|
||||
LOG.debug('returning from RpcWorker wait()')
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def stop(self):
|
||||
LOG.debug('calling RpcWorker stop()')
|
||||
for server in self._servers:
|
||||
if isinstance(server, rpc_server.MessageHandlingServer):
|
||||
LOG.debug('calling stop on %s', server)
|
||||
server.stop()
|
||||
|
||||
@staticmethod
|
||||
def reset():
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def reset(self):
|
||||
config.reset_service()
|
||||
|
||||
|
||||
@@ -190,28 +191,34 @@ def _get_ovn_maintenance_worker():
|
||||
|
||||
|
||||
class AllServicesNeutronWorker(neutron_worker.NeutronBaseWorker):
|
||||
desc = 'services worker'
|
||||
|
||||
def __init__(self, services, worker_process_count=1):
|
||||
super().__init__(worker_process_count)
|
||||
super().__init__(worker_process_count, desc=self.desc)
|
||||
self._services = services
|
||||
for srv in self._services:
|
||||
self._check_base_worker_service(srv)
|
||||
self._launcher = common_service.Launcher(cfg.CONF,
|
||||
restart_method='mutate')
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def start(self):
|
||||
for srv in self._services:
|
||||
# Unset the 'set_proctitle' flag to prevent each service to
|
||||
# re-write the process title already defined and set by this class.
|
||||
srv.set_proctitle = 'off'
|
||||
self._launcher.launch_service(srv)
|
||||
super().start(desc="services worker")
|
||||
super().start(desc=self.desc)
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def stop(self):
|
||||
self._launcher.stop()
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def wait(self):
|
||||
self._launcher.wait()
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def reset(self):
|
||||
self._launcher.restart()
|
||||
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from neutron.common import utils as common_utils
|
||||
from neutron.services.bgp import reconciler
|
||||
from neutron import worker
|
||||
|
||||
@@ -20,17 +21,22 @@ from neutron import worker
|
||||
class BGPWorker(worker.NeutronBaseWorker):
|
||||
def __init__(self):
|
||||
self._reconciler = reconciler.BGPTopologyReconciler()
|
||||
super().__init__(worker_process_count=0)
|
||||
super().__init__(worker_process_count=0, desc="bgp worker")
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def start(self):
|
||||
super().start(desc="bgp worker")
|
||||
super().start(desc=self.desc)
|
||||
self._reconciler.start()
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def wait(self):
|
||||
self._reconciler.full_sync()
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def stop(self):
|
||||
self._reconciler.stop()
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc,
|
||||
finished_only=True)
|
||||
def reset(self):
|
||||
pass
|
||||
@@ -187,6 +187,46 @@ class TestExceptionLogger(base.BaseTestCase):
|
||||
lambda: False, timeout=1, exception=FalseException)
|
||||
|
||||
|
||||
class TestLogWorkerLifecycle(base.BaseTestCase):
|
||||
@mock.patch('neutron.tests.unit.common.test_utils.logging.getLogger')
|
||||
def test_logs_called_and_finished(self, get_logger):
|
||||
logger = mock.Mock()
|
||||
get_logger.return_value = logger
|
||||
|
||||
class Worker:
|
||||
desc = 'test worker'
|
||||
|
||||
@utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def start(self):
|
||||
return 'ok'
|
||||
|
||||
self.assertEqual('ok', Worker().start())
|
||||
logger.debug.assert_has_calls([
|
||||
mock.call('%s %s called, description: %s',
|
||||
'Worker', 'start', 'test worker'),
|
||||
mock.call('%s %s finished, description: %s',
|
||||
'Worker', 'start', 'test worker'),
|
||||
])
|
||||
|
||||
@mock.patch('neutron.tests.unit.common.test_utils.logging.getLogger')
|
||||
def test_finished_only(self, get_logger):
|
||||
logger = mock.Mock()
|
||||
get_logger.return_value = logger
|
||||
|
||||
class Worker:
|
||||
desc = 'test worker'
|
||||
|
||||
@utils.log_worker_lifecycle(lambda self: self.desc,
|
||||
finished_only=True)
|
||||
def reset(self):
|
||||
pass
|
||||
|
||||
Worker().reset()
|
||||
logger.debug.assert_called_once_with(
|
||||
'%s %s finished, description: %s',
|
||||
'Worker', 'reset', 'test worker')
|
||||
|
||||
|
||||
class TestDvrServices(base.BaseTestCase):
|
||||
|
||||
def _test_is_dvr_serviced(self, device_owner, expected):
|
||||
|
||||
@@ -14,6 +14,8 @@ from neutron_lib import worker
|
||||
from oslo_config import cfg
|
||||
from oslo_service import loopingcall
|
||||
|
||||
from neutron.common import utils as common_utils
|
||||
|
||||
|
||||
class NeutronBaseWorker(worker.BaseWorker):
|
||||
def __init__(self, worker_process_count=1, set_proctitle=None,
|
||||
@@ -40,6 +42,7 @@ class PeriodicWorker(NeutronBaseWorker):
|
||||
self._interval = interval
|
||||
self._initial_delay = initial_delay
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def start(self, **kwargs):
|
||||
super().start()
|
||||
if self._loop is None:
|
||||
@@ -48,14 +51,17 @@ class PeriodicWorker(NeutronBaseWorker):
|
||||
self._loop.start(interval=self._interval,
|
||||
initial_delay=self._initial_delay)
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def wait(self):
|
||||
if self._loop is not None:
|
||||
self._loop.wait()
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def stop(self):
|
||||
if self._loop is not None:
|
||||
self._loop.stop()
|
||||
|
||||
@common_utils.log_worker_lifecycle(lambda self: self.desc)
|
||||
def reset(self):
|
||||
self.stop()
|
||||
self.wait()
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
---
|
||||
other:
|
||||
- |
|
||||
Neutron workers now log DEBUG messages when their lifecycle methods
|
||||
(``start``, ``wait``, ``stop``, and ``reset``) are called and finished.
|
||||
The log entries include the worker class name and worker description,
|
||||
which helps troubleshoot ``neutron-periodic-workers``, RPC workers, and
|
||||
other child processes.
|
||||
Reference in new issue
Block a user