Merge pull request #1163 from tcely/patch-2

Write `TaskHistory` rows

Author: meeb

tubesync/common/huey.py

@@ -1,3 +1,4 @@
+import datetime
 import os
 from functools import wraps
 from huey import (
@@ -198,9 +199,13 @@ def on_interrupted(signal_name, task_obj, exception_obj=None, /, *, huey=None):
     assert hasattr(huey, 'enqueue') and callable(huey.enqueue)
     huey.enqueue(task_obj)
 
+storage_key_prefix = 'task_history:'
+
 def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None):
     signal_time = utils.time_clock()
+    signal_dt = datetime.datetime.now(datetime.timezone.utc)
 
+    from common.models import TaskHistory
     add_to_elapsed_signals = frozenset((
         signals.SIGNAL_INTERRUPTED,
         signals.SIGNAL_ERROR,
@@ -214,7 +219,7 @@ def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None):
         signals.SIGNAL_EXECUTING,
         signals.SIGNAL_RETRYING,
     )) | add_to_elapsed_signals
-    storage_key = f'task_history:{task_obj.id}'
+    storage_key = f'{storage_key_prefix}{task_obj.id}'
     task_obj_attr = '_signals_history'
 
     history = getattr(task_obj, task_obj_attr, None)
@@ -224,13 +229,15 @@ def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None):
             key=storage_key,
             peek=True,
         ) or dict(
+            created=signal_dt,
             data=task_obj.data,
             elapsed=0,
             module=task_obj.__module__,
             name=task_obj.name,
         )
         setattr(task_obj, task_obj_attr, history)
     assert history is not None
+    history['modified'] = signal_dt
 
     if signal_name in recorded_signals:
         history[signal_name] = signal_time
@@ -240,12 +247,61 @@ def historical_task(signal_name, task_obj, exception_obj=None, /, *, huey=None):
         huey.get(key=storage_key)
     else:
         huey.put(key=storage_key, data=history)
+    th, created = TaskHistory.objects.get_or_create(
+        task_id=str(task_obj.id),
+        name=f"{task_obj.__module__}.{task_obj.name}",
+        queue=huey.name,
+    )
+    th.priority = task_obj.priority
+    th.task_params = list((
+        list(task_obj.args),
+        repr(task_obj.kwargs),
+    ))
+    if signal_name == signals.SIGNAL_EXECUTING:
+        th.attempts += 1
+        th.start_at = signal_dt
+    elif exception_obj is not None:
+        th.failed_at = signal_dt
+        th.last_error = str(exception_obj)
+    elif signal_name == signals.SIGNAL_ENQUEUED:
+        from sync.models import Media, Source
+        if not th.verbose_name and task_obj.args:
+            key = task_obj.args[0]
+            for model in (Media, Source,):
+                try:
+                    model_instance = model.objects.get(pk=key)
+                except model.DoesNotExist:
+                    pass
+                else:
+                    if hasattr(model_instance, 'key'):
+                        th.verbose_name = f'{th.name} with: {model_instance.key}'
+                        if hasattr(model_instance, 'name'):
+                            th.verbose_name += f' / {model_instance.name}'
+    th.end_at = signal_dt
+    th.save()
 
 # Registration of shared signal handlers
 
 def register_huey_signals():
-    from django_huey import DJANGO_HUEY, signal
+    from django_huey import DJANGO_HUEY, get_queue, signal
     for qn in DJANGO_HUEY.get('queues', dict()):
         signal(signals.SIGNAL_INTERRUPTED, queue=qn)(on_interrupted)
         signal(queue=qn)(historical_task)
 
+        # clean up old history and results from storage
+        q = get_queue(qn)
+        now_time = utils.time_clock()
+        for key in q.all_results().keys():
+            if not key.startswith(storage_key_prefix):
+                continue
+            history = q.get(peek=True, key=key)
+            if not isinstance(history, dict):
+                continue
+            age = datetime.timedelta(
+                seconds=(now_time - history.get(signals.SIGNAL_EXECUTING, now_time)),
+            )
+            if age > datetime.timedelta(days=7):
+                result_key = key[len(storage_key_prefix) :]
+                q.get(peek=False, key=result_key)
+                q.get(peek=False, key=key)
+

tubesync/sync/signals.py

@@ -6,10 +6,12 @@
 from django.db.models.signals import pre_save, post_save, pre_delete, post_delete
 from django.db.transaction import atomic, on_commit
 from django.dispatch import receiver
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
-from background_task.signals import task_failed
+from background_task.signals import task_started, task_successful, task_failed
 from background_task.models import Task
 from common.logger import log
+from common.models import TaskHistory
 from common.utils import glob_quote, mkdir_p
 from .models import Source, Media, Metadata
 from .tasks import (
@@ -166,8 +168,55 @@ def source_post_delete(sender, instance, **kwargs):
     delete_task_by_source('sync.tasks.save_all_media_for_source', instance.pk)
 
 
-@receiver(task_failed, sender=Task)
+@receiver(task_started, dispatch_uid='sync.signals.task_task_started')
+@atomic(durable=False)
+def task_task_started(sender, **kwargs):
+    locked_tasks = Task.objects.locked(timezone.now())
+    for task_obj in locked_tasks:
+        th, created = TaskHistory.objects.get_or_create(
+            task_id=str(task_obj.pk),
+            name=task_obj.task_name,
+            queue=task_obj.queue,
+        )
+        th.attempts += 1
+        th.end_at = task_obj.locked_at
+        th.priority = (100 - task_obj.priority)
+        th.repeat = task_obj.repeat
+        th.repeat_until = task_obj.repeat_until
+        th.start_at = task_obj.locked_at
+        th.task_params = list(task_obj.params())
+        th.verbose_name = task_obj.verbose_name
+        th.save()
+        if created:
+            log.debug(f'Created a new task history record: {th.pk}: {th.verbose_name}')
+
+
+def merge_completed_task_into_history(task_id, task_obj):
+    th, created = TaskHistory.objects.get_or_create(
+        task_id=str(task_id),
+        name=task_obj.task_name,
+        queue=task_obj.queue,
+    )
+    th.end_at = task_obj.run_at
+    th.failed_at = task_obj.failed_at
+    th.last_error = task_obj.last_error
+    th.repeat = task_obj.repeat
+    th.repeat_until = task_obj.repeat_until
+    th.start_at = task_obj.locked_at
+    th.verbose_name = task_obj.verbose_name
+    th.save()
+
+
+@receiver(task_successful, dispatch_uid='sync.signals.task_task_successful')
+@atomic(durable=False)
+def task_task_successful(sender, task_id, completed_task, **kwargs):
+    merge_completed_task_into_history(task_id, completed_task)
+
+
+@receiver(task_failed, dispatch_uid='sync.signals.task_task_failed')
+@atomic(durable=False)
 def task_task_failed(sender, task_id, completed_task, **kwargs):
+    merge_completed_task_into_history(task_id, completed_task)
     # Triggered after a task fails by reaching its max retry attempts
     obj, url = map_task_to_instance(completed_task)
     if isinstance(obj, Source):

tubesync/sync/tasks.py

@@ -27,6 +27,7 @@
 from huey import crontab as huey_crontab, signals as huey_signals
 from common.huey import CancelExecution, dynamic_retry, register_huey_signals
 from common.logger import log
+from common.models import TaskHistory
 from common.errors import ( BgTaskWorkerError, DownloadFailedException,
                             NoFormatException, NoMediaException,
                             NoThumbnailException, )
@@ -199,6 +200,7 @@ def cleanup_completed_tasks():
     log.info(f'Deleting completed tasks older than {days_to_keep} days '
              f'(run_at before {delta})')
     CompletedTask.objects.filter(run_at__lt=delta).delete()
+    TaskHistory.objects.filter(end_at__lt=delta).delete()
 
 
 @atomic(durable=False)