implement job ids, fix #53

Author: samuelcolvin

HISTORY.rst

@@ -3,10 +3,11 @@
 History
 -------
 
-v0.9.1 (2017-08-16)
-...................
+v0.10.0 (2017-08-16)
+....................
 * log redis version when starting worker, fix #64
 * log "connection success" when connecting to redis after connection failures, fix #67
+* add job ids, for now they're just used in logging, fix #53
 
 v0.9.0 (2017-06-23)
 ...................

arq/jobs.py

@@ -4,6 +4,8 @@
 
 Defines the ``Job`` class and descendants which deal with encoding and decoding job data.
 """
+import base64
+import os
 from datetime import datetime
 
 import msgpack
@@ -21,6 +23,15 @@ class JobSerialisationError(ArqError):
     pass
 
 
+def gen_random():
+    """
+    generate a lowercase alpha-numeric random string of length 24.
+
+    Should have more randomness for its size thank uuid
+    """
+    return base64.b32encode(os.urandom(10))[:16].decode().lower()
+
+
 # "device control one" should be fairly unique as a dict key and only one byte
 DEVICE_CONTROL_ONE = '\x11'
 
@@ -30,7 +41,7 @@ class Job:
     Main Job class responsible for encoding and decoding jobs as they go
     into and come out of redis.
     """
-    __slots__ = ('queue', 'queued_at', 'class_name', 'func_name', 'args', 'kwargs', 'raw_queue', 'raw_data')
+    __slots__ = 'id', 'queue', 'queued_at', 'class_name', 'func_name', 'args', 'kwargs', 'raw_queue', 'raw_data'
 
     def __init__(self, raw_data: bytes, *, queue_name: str=None, raw_queue: bytes=None) -> None:
         """
@@ -45,16 +56,17 @@ def __init__(self, raw_data: bytes, *, queue_name: str=None, raw_queue: bytes=No
             raise ArqError('either queue_name or raw_queue are required')
         self.queue = queue_name or raw_queue.decode()
         self.raw_queue = raw_queue or queue_name.encode()
-        self.queued_at, self.class_name, self.func_name, self.args, self.kwargs = self._decode(raw_data)
+        self.queued_at, self.class_name, self.func_name, self.args, self.kwargs, self.id = self.decode_raw(raw_data)
         self.queued_at /= 1000
 
     @classmethod
-    def encode(cls, *, queued_at: int=None, class_name: str, func_name: str,
+    def encode(cls, *, job_id: str=None, queued_at: int=None, class_name: str, func_name: str,
                args: tuple, kwargs: dict) -> bytes:
         """
         Create a byte string suitable for pushing into redis which contains all
         required information about a job to be performed.
 
+        :param job_id: id to use for the job, leave blank to generate a uuid
         :param queued_at: time in ms unix time when the job was queue, if None now is used
         :param class_name: name (see :attr:`arq.main.Actor.name`) of the actor class where the job is defined
         :param func_name: name of the function be called
@@ -63,10 +75,14 @@ def encode(cls, *, queued_at: int=None, class_name: str, func_name: str,
         """
         queued_at = queued_at or int(timestamp() * 1000)
         try:
-            return cls._encode([queued_at, class_name, func_name, args, kwargs])
+            return cls.encode_raw([queued_at, class_name, func_name, args, kwargs, cls.generate_id(job_id)])
         except TypeError as e:
             raise JobSerialisationError(str(e)) from e
 
+    @classmethod
+    def generate_id(cls, given_id):
+        return given_id or gen_random()
+
     @classmethod
     def msgpack_encoder(cls, obj):
         """
@@ -84,11 +100,11 @@ def msgpack_object_hook(cls, obj):
         return obj
 
     @classmethod
-    def _encode(cls, data) -> bytes:
+    def encode_raw(cls, data) -> bytes:
         return msgpack.packb(data, default=cls.msgpack_encoder, use_bin_type=True)
 
     @classmethod
-    def _decode(cls, data: bytes):
+    def decode_raw(cls, data: bytes):
         return msgpack.unpackb(data, object_hook=cls.msgpack_object_hook, encoding='utf8')
 
     def to_string(self, args_curtail=DEFAULT_CURTAIL):
@@ -100,7 +116,10 @@ def to_string(self, args_curtail=DEFAULT_CURTAIL):
                 arguments += ', '
             arguments += ', '.join(f'{k}={v!r}' for k, v in sorted(self.kwargs.items()))
 
-        return '{s.class_name}.{s.func_name}({args})'.format(s=self, args=truncate(arguments, args_curtail))
+        return '{s.id:.6} {s.class_name}.{s.func_name}({args})'.format(s=self, args=truncate(arguments, args_curtail))
+
+    def short_ref(self):
+        return '{s.id:.6} {s.class_name}.{s.func_name}'.format(s=self)
 
     def __str__(self):
         return self.to_string()

arq/version.py

@@ -2,4 +2,4 @@
 
 __all__ = ['VERSION']
 
-VERSION = StrictVersion('0.9.1')
+VERSION = StrictVersion('0.10.0')

arq/worker.py

@@ -330,7 +330,7 @@ def log_job_result(self, started_at: float, result, j: Job):
             return
         job_time = timestamp() - started_at
         sr = '' if result is None else truncate(repr(result), self.log_curtail)
-        jobs_logger.info('%-4s ran in%7.3fs ← %s.%s ● %s', j.queue, job_time, j.class_name, j.func_name, sr)
+        jobs_logger.info('%-4s ran in%7.3fs ← %s ● %s', j.queue, job_time, j.short_ref(), sr)
 
     def handle_prepare_exc(self, msg: str):
         self.drain.jobs_failed += 1
@@ -341,10 +341,10 @@ def handle_prepare_exc(self, msg: str):
     @classmethod
     def handle_stop_job(cls, started_at: float, exc: StopJob, j: Job):
         if exc.warning:
-            msg, logger = '%-4s ran in%7.3fs ■ %s.%s ● Stopped Warning %s', jobs_logger.warning
+            msg, logger = '%-4s ran in%7.3fs ■ %s ● Stopped Warning %s', jobs_logger.warning
         else:
-            msg, logger = '%-4s ran in%7.3fs ■ %s.%s ● Stopped %s', jobs_logger.info
-        logger(msg, j.queue, timestamp() - started_at, j.class_name, j.func_name, exc)
+            msg, logger = '%-4s ran in%7.3fs ■ %s ● Stopped %s', jobs_logger.info
+        logger(msg, j.queue, timestamp() - started_at, j.short_ref(), exc)
 
     @classmethod
     def handle_execute_exc(cls, started_at: float, exc: BaseException, j: Job):

tests/fixtures.py

@@ -6,12 +6,20 @@
 import time
 from pathlib import Path
 
-from arq import Actor, BaseWorker, StopJob, concurrent, cron
+from arq import Actor, BaseWorker, Job, StopJob, concurrent, cron
 from arq.drain import Drain
 from arq.testing import MockRedisMixin
 
 
+class JobConstID(Job):
+    @classmethod
+    def generate_id(cls, given_id):
+        return '__id__'
+
+
 class DemoActor(Actor):
+    job_class = JobConstID
+
     @concurrent
     async def add_numbers(self, a, b):
         """add_number docs"""
@@ -88,6 +96,10 @@ async def stop_job_warning(self):
         raise StopJob('stopping job with warning', warning=True)
 
 
+class RealJobActor(DemoActor):
+    job_class = Job
+
+
 class MockRedisDemoActor(MockRedisMixin, DemoActor):
     pass
 
@@ -97,6 +109,8 @@ class Worker(BaseWorker):
 
 
 class StartupActor(Actor):
+    job_class = JobConstID
+
     async def startup(self):
         with open('events', 'a') as f:
             f.write('startup[{}],'.format(self.is_shadow))
@@ -190,6 +204,7 @@ class ReEnqueueActor(DemoActor):
 
 class CronActor(Actor):
     # using 3:0:0 makes it very unlikely the job will be caused due hitting the right time
+    job_class = JobConstID
 
     @cron(hour=3, minute=0, second=0, run_at_startup=True)
     async def save_foobar(self):

tests/test_customisation.py

@@ -7,7 +7,7 @@
 from arq.jobs import DatetimeJob, Job, JobSerialisationError
 from arq.worker import BaseWorker
 
-from .fixtures import DemoActor
+from .fixtures import DemoActor, RealJobActor
 
 
 class DatetimeActor(DemoActor):
@@ -77,7 +77,7 @@ async def test_encode_non_datetimes(tmpworkdir, loop, redis_conn):
 
 
 async def test_wrong_job_class(loop):
-    worker = DatetimeWorker(loop=loop, burst=True, shadows=[DemoActor, DemoActor, DatetimeActor])
+    worker = DatetimeWorker(loop=loop, burst=True, shadows=[RealJobActor, RealJobActor, DatetimeActor])
     with pytest.raises(TypeError) as excinfo:
         await worker.run()
     assert excinfo.value.args[0].endswith("has a different job class to the first shadow, "
@@ -99,7 +99,7 @@ class DifferentQueuesActor(DemoActor):
 
 
 async def test_switch_job_class(loop):
-    worker = DatetimeWorker(loop=loop, burst=True, shadows=[DemoActor])
+    worker = DatetimeWorker(loop=loop, burst=True, shadows=[RealJobActor])
     assert worker.job_class is None
     await worker.run()
     assert worker.job_class == Job
@@ -109,26 +109,26 @@ async def test_switch_job_class(loop):
 def test_naïve_dt_encoding():
     t = datetime(2000, 1, 1)
     assert str(t) == '2000-01-01 00:00:00'
-    p = DatetimeJob._encode(t)
-    t2 = DatetimeJob._decode(p)
+    p = DatetimeJob.encode_raw(t)
+    t2 = DatetimeJob.decode_raw(p)
     assert t == t2
     assert str(t2) == '2000-01-01 00:00:00'
 
 
 def test_utc_dt_encoding():
     t = datetime(2000, 1, 1, tzinfo=timezone.utc)
     assert str(t) == '2000-01-01 00:00:00+00:00'
-    p = DatetimeJob._encode(t)
-    t2 = DatetimeJob._decode(p)
+    p = DatetimeJob.encode_raw(t)
+    t2 = DatetimeJob.decode_raw(p)
     assert t == t2
     assert str(t2) == '2000-01-01 00:00:00+00:00'
 
 
 def test_new_york_dt_encoding():
     t = datetime(2000, 1, 1, tzinfo=timezone(timedelta(hours=-5)))
     assert str(t) == '2000-01-01 00:00:00-05:00'
-    p = DatetimeJob._encode(t)
-    t2 = DatetimeJob._decode(p)
+    p = DatetimeJob.encode_raw(t)
+    t2 = DatetimeJob.decode_raw(p)
     assert t == t2
     assert str(t2) == '2000-01-01 00:00:00-05:00'
 
@@ -137,8 +137,8 @@ def test_pytz_new_york_dt_encoding():
     ny = pytz.timezone('America/New_York')
     t = ny.localize(datetime(2000, 1, 1))
     assert str(t) == '2000-01-01 00:00:00-05:00'
-    p = DatetimeJob._encode(t)
-    t2 = DatetimeJob._decode(p)
+    p = DatetimeJob.encode_raw(t)
+    t2 = DatetimeJob.decode_raw(p)
     assert t == t2
     assert datetime(2000, 1, 1, tzinfo=timezone(timedelta(hours=-5))) == t2
     assert str(t2) == '2000-01-01 00:00:00-05:00'
@@ -147,17 +147,17 @@ def test_pytz_new_york_dt_encoding():
 def test_dt_encoding_with_ms():
     t = datetime(2000, 1, 1, 0, 0, 0, 123000)
     assert str(t) == '2000-01-01 00:00:00.123000'
-    p = DatetimeJob._encode(t)
-    t2 = DatetimeJob._decode(p)
+    p = DatetimeJob.encode_raw(t)
+    t2 = DatetimeJob.decode_raw(p)
     assert t == t2
     assert str(t2) == '2000-01-01 00:00:00.123000'
 
 
 def test_dt_encoding_with_μs():
     t = datetime(2000, 1, 1, 0, 0, 0, 123456)
     assert str(t) == '2000-01-01 00:00:00.123456'
-    p = DatetimeJob._encode(t)
-    t2 = DatetimeJob._decode(p)
+    p = DatetimeJob.encode_raw(t)
+    t2 = DatetimeJob.decode_raw(p)
     assert t != t2
     assert (t - t2) == timedelta(microseconds=456)
     assert str(t2) == '2000-01-01 00:00:00.123000'

tests/test_doc_example.py

@@ -13,5 +13,9 @@ async def test_run_job_burst(redis_conn, loop, caplog):
     await downloader.download_content('http://example.com')
     await worker.run()
     await downloader.close()
-    assert 's → Downloader.download_content(http://example.com)' in caplog
-    assert 's ← Downloader.download_content ● 1' in caplog
+    log = caplog(
+        (' [0-9a-z]{6} ', ' __id__ ')
+    )
+    print(log)
+    assert 's → __id__ Downloader.download_content(http://example.com)' in log
+    assert 's ← __id__ Downloader.download_content ● 1' in log

tests/test_main.py

@@ -24,7 +24,7 @@ async def test_simple_job_dispatch(tmpworkdir, loop, debug):
     data = msgpack.unpackb(v[0], encoding='utf8')
     # timestamp
     assert 1e12 < data.pop(0) < 3e12
-    assert data == ['MockRedisDemoActor', 'add_numbers', [1, 2], {}]
+    assert data == ['MockRedisDemoActor', 'add_numbers', [1, 2], {}, '__id__']
 
 
 async def test_concurrency_disabled_job_dispatch(tmpworkdir, loop):
@@ -44,7 +44,7 @@ async def test_enqueue_redis_job(actor, redis_conn):
     data = msgpack.unpackb(dft_queue[0], encoding='utf8')
     # timestamp
     assert 1e12 < data.pop(0) < 3e12
-    assert data == ['DemoActor', 'add_numbers', [1, 2], {}]
+    assert data == ['DemoActor', 'add_numbers', [1, 2], {}, '__id__']
 
 
 async def test_dispatch_work(tmpworkdir, loop, caplog, redis_conn):
@@ -69,22 +69,22 @@ async def test_dispatch_work(tmpworkdir, loop, caplog, redis_conn):
     assert ('MockRedisDemoActor.add_numbers → dft\n'
             'MockRedisDemoActor.high_add_numbers → high\n'
             'Initialising work manager, burst mode: True, creating shadows...\n'
-            'Using first shadows job class "Job"\n'
+            'Using first shadows job class "JobConstID"\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=1 q_dft=1 q_low=0\n'
             'starting main blpop loop\n'
             'populating quit queue to prompt exit: arq:quit-<random>\n'
             'jobs in progress 1\n'
-            'scheduling job <Job MockRedisDemoActor.high_add_numbers(3, 4, c=5) on high>, re-enqueue: False\n'
+            'scheduling job <Job __id__ MockRedisDemoActor.high_add_numbers(3, 4, c=5) on high>, re-enqueue: False\n'
             'jobs in progress 2\n'
-            'scheduling job <Job MockRedisDemoActor.add_numbers(1, 2) on dft>, re-enqueue: False\n'
+            'scheduling job <Job __id__ MockRedisDemoActor.add_numbers(1, 2) on dft>, re-enqueue: False\n'
             'got job from the quit queue, stopping\n'
             'drain waiting 5.0s for 2 tasks to finish\n'
-            'high queued  0.0XXs → MockRedisDemoActor.high_add_numbers(3, 4, c=5)\n'
-            'high ran in  0.0XXs ← MockRedisDemoActor.high_add_numbers ● 12\n'
-            'dft  queued  0.0XXs → MockRedisDemoActor.add_numbers(1, 2)\n'
-            'dft  ran in  0.0XXs ← MockRedisDemoActor.add_numbers ● \n'
+            'high queued  0.0XXs → __id__ MockRedisDemoActor.high_add_numbers(3, 4, c=5)\n'
+            'high ran in  0.0XXs ← __id__ MockRedisDemoActor.high_add_numbers ● 12\n'
+            'dft  queued  0.0XXs → __id__ MockRedisDemoActor.add_numbers(1, 2)\n'
+            'dft  ran in  0.0XXs ← __id__ MockRedisDemoActor.add_numbers ● \n'
             'task complete, 1 jobs done, 0 failed\n'
             'task complete, 2 jobs done, 0 failed\n'
             'shutting down worker after 0.0XXs ◆ 2 jobs done ◆ 0 failed ◆ 0 timed out\n') == log
@@ -109,8 +109,8 @@ async def test_handle_exception(loop, caplog):
             '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'
             'drain waiting 5.0s for 1 tasks to finish\n'
-            'dft  queued  0.0XXs → MockRedisDemoActor.boom()\n'
-            'dft  ran in  0.0XXs ! MockRedisDemoActor.boom(): RuntimeError\n'
+            'dft  queued  0.0XXs → __id__ MockRedisDemoActor.boom()\n'
+            'dft  ran in  0.0XXs ! __id__ MockRedisDemoActor.boom(): RuntimeError\n'
             'Traceback (most recent call last):\n'
             '  File "/path/to/arq/worker.py", line <no>, in run_job\n'
             '    result = await func(*j.args, **j.kwargs)\n'
@@ -157,8 +157,8 @@ async def test_call_direct(mock_actor_worker, caplog):
     assert worker.jobs_failed == 0
     assert worker.jobs_complete == 1
     log = re.sub('0.0\d\ds', '0.0XXs', caplog.log)
-    assert ('arq.jobs: dft  queued  0.0XXs → MockRedisDemoActor.direct_method(1, 2)\n'
-            'arq.jobs: dft  ran in  0.0XXs ← MockRedisDemoActor.direct_method ● 3') in log
+    assert ('arq.jobs: dft  queued  0.0XXs → __id__ MockRedisDemoActor.direct_method(1, 2)\n'
+            'arq.jobs: dft  ran in  0.0XXs ← __id__ MockRedisDemoActor.direct_method ● 3') in log
 
 
 async def test_direct_binding(mock_actor_worker, caplog):