Wait until workers have been launched

In "test_restart_rpc_on_sighup_multiple_workers", the test needs to
wait until the RPC workers have been properly launched by
``oslo_service.service.ProcessLauncher.launch_service``. Once this
method returns, it is guaranteed that the child worker processes
are running and the signal process handlers are attending the
SIGHUP signal that will reset them.

Closes-Bug: #1938428
Change-Id: I1dc56092d099223accc3aefa8e303310c4f6787e
This commit is contained in:
Rodolfo Alonso Hernandez 2021-07-29 10:30:01 +00:00
parent 0411743bb9
commit fafcabdbe0
1 changed files with 34 additions and 22 deletions

View File

@ -15,6 +15,7 @@
import multiprocessing
import os
import queue
import signal
import socket
import time
@ -62,7 +63,7 @@ class TestNeutronServer(base.BaseLoggingTestCase):
# Make sure all processes are stopped
os.kill(self.service_pid, signal.SIGKILL)
def _start_server(self, callback, workers):
def _start_server(self, callback, workers, processes_queue=None):
"""Run a given service.
:param callback: callback that will start the required service
@ -92,15 +93,7 @@ class TestNeutronServer(base.BaseLoggingTestCase):
# If number of workers is 1 it is assumed that we run
# a service in the current process.
if self.workers > 1:
# Wait at most 10 seconds to spawn workers
condition = lambda: self.workers == len(self._get_workers())
utils.wait_until_true(
condition, timeout=10, sleep=0.1,
exception=RuntimeError(
"Failed to start %d workers." % self.workers))
workers = self._get_workers()
workers = self._get_workers(10, processes_queue=processes_queue)
self.assertEqual(len(workers), self.workers)
return workers
@ -111,21 +104,35 @@ class TestNeutronServer(base.BaseLoggingTestCase):
return [self.service_pid]
def _get_workers(self):
def _get_workers(self, timeout, processes_queue=None):
"""Get the list of processes in which WSGI server is running."""
def safe_ppid(proc):
try:
return proc.ppid()
except psutil.NoSuchProcess:
return None
if self.workers > 1:
return [proc.pid for proc in psutil.process_iter()
if safe_ppid(proc) == self.service_pid]
else:
return [proc.pid for proc in psutil.process_iter()
if proc.pid == self.service_pid]
def get_workers_pid():
if self.workers > 1:
return [proc.pid for proc in psutil.process_iter()
if safe_ppid(proc) == self.service_pid]
else:
return [proc.pid for proc in psutil.process_iter()
if proc.pid == self.service_pid]
exception = RuntimeError('Failed to start %d workers.' % self.workers)
if processes_queue:
try:
return processes_queue.get(timeout=timeout)
except queue.Empty:
raise exception
# Wait at most 10 seconds to spawn workers
condition = lambda: self.workers == len(get_workers_pid())
utils.wait_until_true(condition, timeout=timeout, sleep=0.1,
exception=exception)
return get_workers_pid()
def _check_active(self):
"""Dummy service activity check."""
@ -138,7 +145,8 @@ class TestNeutronServer(base.BaseLoggingTestCase):
def _fake_reset(self):
self._mp_queue.put(FAKE_RESET_MSG)
def _test_restart_service_on_sighup(self, service, workers=1):
def _test_restart_service_on_sighup(self, service, workers=1,
processes_queue=None):
"""Test that a service correctly (re)starts on receiving SIGHUP.
1. Start a service with a given number of workers.
@ -146,7 +154,8 @@ class TestNeutronServer(base.BaseLoggingTestCase):
3. Wait for workers (if any) to (re)start.
"""
self._start_server(callback=service, workers=workers)
self._start_server(callback=service, workers=workers,
processes_queue=processes_queue)
os.kill(self.service_pid, signal.SIGHUP)
# After sending SIGHUP it is expected that there will be as many
@ -246,6 +255,7 @@ class TestRPCServer(TestNeutronServer):
self._plugin_patcher = mock.patch(TARGET_PLUGIN, autospec=True)
self.plugin = self._plugin_patcher.start()
self.plugin.return_value.rpc_workers_supported = True
self._processes_queue = multiprocessing.Queue()
def _serve_rpc(self, workers=1):
"""Start RPC server with a given number of workers."""
@ -266,11 +276,13 @@ class TestRPCServer(TestNeutronServer):
CONF.set_override("rpc_state_report_workers", 0)
rpc_workers_launcher = service.start_rpc_workers()
self._processes_queue.put(list(rpc_workers_launcher.children))
rpc_workers_launcher.wait()
def test_restart_rpc_on_sighup_multiple_workers(self):
self._test_restart_service_on_sighup(service=self._serve_rpc,
workers=2)
self._test_restart_service_on_sighup(
service=self._serve_rpc, workers=2,
processes_queue=self._processes_queue)
class TestPluginWorker(TestNeutronServer):