adding async-timeout

Author: samuelcolvin

HISTORY.rst

@@ -3,6 +3,11 @@
 History
 -------
 
+v0.8.0 (TBC)
+............
+* add ``async-timeout`` dependency and use async timeout around ``shadow_factory``
+* change logger name for control process log messages
+
 v0.7.0 (2017-06-01)
 ...................
 * implementing reusable ``Drain`` which takes tasks from a redis list and allows them to be execute asynchronously.

Makefile

@@ -1,3 +1,5 @@
+.DEFAULT_GOAL := all
+
 .PHONY: install
 install:
 	pip install -U pip setuptools

arq/version.py

@@ -2,4 +2,4 @@
 
 __all__ = ['VERSION']
 
-VERSION = StrictVersion('0.7.0')
+VERSION = StrictVersion('0.8.0')

arq/worker.py

@@ -16,6 +16,8 @@
 from signal import Signals
 from typing import Dict, List, Type  # noqa
 
+from async_timeout import timeout
+
 from .drain import Drain
 from .jobs import ArqError, Job
 from .logs import default_log_config
@@ -25,6 +27,7 @@
 __all__ = ['BaseWorker', 'RunWorkerProcess', 'StopJob', 'import_string']
 
 work_logger = logging.getLogger('arq.work')
+ctrl_logger = logging.getLogger('arq.control')
 jobs_logger = logging.getLogger('arq.jobs')
 
 
@@ -173,9 +176,10 @@ async def run(self):
         perform jobs.
         """
         self._stopped = False
-        work_logger.info('Initialising work manager, burst mode: %s', self._burst_mode)
+        work_logger.info('Initialising work manager, burst mode: %s, creating shadows...', self._burst_mode)
 
-        shadows = await self.shadow_factory()
+        with timeout(10):
+            shadows = await self.shadow_factory()
         assert isinstance(shadows, list), 'shadow_factory should return a list not %s' % type(shadows)
         self.job_class = shadows[0].job_class
         work_logger.debug('Using first shadows job class "%s"', self.job_class.__name__)
@@ -416,8 +420,8 @@ def start_worker(worker_path: str, worker_class: str, burst: bool, loop: asyncio
     :param loop: asyncio loop use to or None
     """
     worker_cls = import_string(worker_path, worker_class)
-    worker = worker_cls(burst=burst, loop=loop)
     work_logger.info('Starting "%s" on pid=%d', worker_cls.__name__, os.getpid())
+    worker = worker_cls(burst=burst, loop=loop)
     try:
         worker.run_until_complete()
     except HandledExit:
@@ -444,32 +448,32 @@ def __init__(self, worker_path, worker_class, burst=False):
 
     def run_worker(self, worker_path, worker_class, burst):
         name = 'WorkProcess'
-        work_logger.info('starting work process "%s"', name)
+        ctrl_logger.info('starting work process "%s"', name)
         self.process = Process(target=start_worker, args=(worker_path, worker_class, burst), name=name)
         self.process.start()
         self.process.join()
         if self.process.exitcode == 0:
-            work_logger.info('worker process exited ok')
+            ctrl_logger.info('worker process exited ok')
             return
-        work_logger.critical('worker process %s exited badly with exit code %s',
+        ctrl_logger.critical('worker process %s exited badly with exit code %s',
                              self.process.pid, self.process.exitcode)
         sys.exit(3)
         # could restart worker here, but better to leave it up to the real manager eg. docker restart: always
 
     def handle_sig(self, signum, frame):
         signal.signal(signal.SIGINT, self.handle_sig_force)
         signal.signal(signal.SIGTERM, self.handle_sig_force)
-        work_logger.info('got signal: %s, waiting for worker pid=%s to finish...', Signals(signum).name,
+        ctrl_logger.info('got signal: %s, waiting for worker pid=%s to finish...', Signals(signum).name,
                          self.process and self.process.pid)
         # sleep to make sure worker.handle_sig above has executed if it's going to and detached handle_proxy_signal
         time.sleep(0.01)
         if self.process and self.process.is_alive():
-            work_logger.debug("sending custom shutdown signal to worker in case it didn't receive the signal")
+            ctrl_logger.debug("sending custom shutdown signal to worker in case it didn't receive the signal")
             os.kill(self.process.pid, SIG_PROXY)
 
     def handle_sig_force(self, signum, frame):
-        work_logger.warning('got signal: %s again, forcing exit', Signals(signum).name)
+        ctrl_logger.warning('got signal: %s again, forcing exit', Signals(signum).name)
         if self.process and self.process.is_alive():
-            work_logger.error('sending worker %d SIGTERM', self.process.pid)
+            ctrl_logger.error('sending worker %d SIGTERM', self.process.pid)
             os.kill(self.process.pid, signal.SIGTERM)
         raise ImmediateExit('force exit')

setup.py

@@ -45,6 +45,7 @@
         arq=arq.cli:cli
     """,
     install_requires=[
+        'async-timeout==1.1.0',
         'aioredis>=0.2.9',
         'click>=6.6',
         'msgpack-python>=0.4.8',

tests/conftest.py

@@ -70,5 +70,5 @@ def mock_actor_worker(mock_actor):
 
 @pytest.fixture
 def caplog(caplog):
-    caplog.set_loggers(log_names=('arq.main', 'arq.work', 'arq.jobs'), fmt='%(name)s: %(message)s')
+    caplog.set_loggers(log_names=('arq.control', 'arq.main', 'arq.work', 'arq.jobs'), fmt='%(name)s: %(message)s')
     return caplog

tests/test_main.py

@@ -68,7 +68,7 @@ async def test_dispatch_work(tmpworkdir, loop, caplog, redis_conn):
     print(log)
     assert ('MockRedisDemoActor.add_numbers ▶ dft\n'
             'MockRedisDemoActor.high_add_numbers ▶ high\n'
-            'Initialising work manager, burst mode: True\n'
+            'Initialising work manager, burst mode: True, creating shadows...\n'
             'Using first shadows job class "Job"\n'
             'Running worker with 1 shadow listening to 3 queues\n'
             'shadows: MockRedisDemoActor | queues: high, dft, low\n'
@@ -102,7 +102,7 @@ async def test_handle_exception(loop, caplog):
     log = re.sub(r'\d{4}-\d+-\d+ \d+:\d+:\d+', '<date time>', log)
     log = re.sub(r'\w{3}-\d+ \d+:\d+:\d+', '<date time2>', log)
     print(log)
-    assert ('Initialising work manager, burst mode: True\n'
+    assert ('Initialising work manager, burst mode: True, creating shadows...\n'
             'Running worker with 1 shadow listening to 3 queues\n'
             'shadows: MockRedisDemoActor | queues: high, dft, low\n'
             'recording health: <date time2> j_complete=0 j_failed=0 j_timedout=0 j_ongoing=0 q_high=0 q_dft=1 q_low=0\n'

tests/test_worker.py

@@ -74,7 +74,7 @@ async def test_separate_log_levels(mock_actor_worker, caplog):
     await actor.concat(a='1', b='2')
     await worker.run()
     log = caplog(('0.0\d\ds', '0.0XXs'))
-    assert ('arq.work: Initialising work manager, burst mode: True\n'
+    assert ('arq.work: Initialising work manager, burst mode: True, creating shadows...\n'
             'arq.work: Running worker with 1 shadow listening to 3 queues\n'
             'arq.work: shadows: MockRedisDemoActor | queues: high, dft, low\n'
             'arq.work: drain waiting 5.0s for 1 tasks to finish\n'