Cancel or delete stale pending and failed exports

.gitignore

@@ -39,6 +39,7 @@ build
 *~
 *orig
 celerybeat-schedule
+celerybeat.pid
 *.swp
 .venv
 *.ropeproject
@@ -72,4 +73,4 @@ bob
 
 # visual studio code
 .vscode/
-tags
+tags

onadata/apps/viewer/tasks.py

@@ -1,8 +1,11 @@
 import sys
+from datetime import timedelta
 
 from celery import task
+from celery.task.schedules import crontab
 from django.conf import settings
 from django.shortcuts import get_object_or_404
+from django.utils import timezone
 from requests import ConnectionError
 
 from onadata.apps.viewer.models.export import Export
@@ -349,3 +352,35 @@ def delete_export(export_id):
         export.delete()
         return True
     return False
+
+
+@task.periodic_task(
+    run_every=(crontab(hour='*/7')),
+    ignore_result=True
+)
+def mark_expired_pending_exports_as_failed():
+    """
+    Exports that have not completed within a set time should be marked as
+    failed
+    """
+    # import pdb; pdb.set_trace()
+    task_lifespan = settings.EXPORT_TASK_LIFESPAN
+    time_threshold = timezone.now() - timedelta(hours=task_lifespan)
+    exports = Export.objects.filter(internal_status=Export.PENDING,
+                                    created_on__lt=time_threshold)
+    exports.update(internal_status=Export.FAILED)
+
+
+@task.periodic_task(
+    run_every=(crontab(hour='*/7')),
+    ignore_result=True
+)
+def delete_expired_failed_exports():
+    """
+    Delete old failed exports
+    """
+    task_lifespan = settings.EXPORT_TASK_LIFESPAN
+    time_threshold = timezone.now() - timedelta(hours=task_lifespan)
+    exports = Export.objects.filter(internal_status=Export.FAILED,
+                                    created_on__lt=time_threshold)
+    exports.delete()

onadata/apps/viewer/tests/test_tasks.py

@@ -1,9 +1,14 @@
+from datetime import timedelta
+
 from celery import current_app
 from django.conf import settings
+from django.utils import timezone
 
 from onadata.apps.main.tests.test_base import TestBase
 from onadata.apps.viewer.models.export import Export
 from onadata.apps.viewer.tasks import create_async_export
+from onadata.apps.viewer.tasks import mark_expired_pending_exports_as_failed
+from onadata.apps.viewer.tasks import delete_expired_failed_exports
 
 
 class TestExportTasks(TestBase):
@@ -33,8 +38,35 @@ def test_create_async(self):
         for export_type, extra_options in export_types:
             result = create_async_export(
                 self.xform, export_type, None, False, options)
-
             export = result[0]
             self.assertTrue(export.id)
             self.assertIn("username", options)
             self.assertEquals(options.get("id_string"), self.xform.id_string)
+
+    def test_mark_expired_pending_exports_as_failed(self):
+        self._publish_transportation_form_and_submit_instance()
+        over_threshold = settings.EXPORT_TASK_LIFESPAN + 2
+        export = Export.objects.create(xform=self.xform,
+                                       export_type=Export.CSV_EXPORT,
+                                       internal_status=Export.PENDING,
+                                       filename="")
+        # we set created_on here because Export.objects.create() overrides it
+        export.created_on = timezone.now() - timedelta(hours=over_threshold)
+        export.save()
+        mark_expired_pending_exports_as_failed()
+        export = Export.objects.filter(pk=export.pk).first()
+        self.assertEquals(export.internal_status, Export.FAILED)
+
+    def test_delete_expired_failed_exports(self):
+        self._publish_transportation_form_and_submit_instance()
+        over_threshold = settings.EXPORT_TASK_LIFESPAN + 2
+        export = Export.objects.create(xform=self.xform,
+                                       export_type=Export.CSV_EXPORT,
+                                       internal_status=Export.FAILED,
+                                       filename="")
+        # we set created_on here because Export.objects.create() overrides it
+        export.created_on = timezone.now() - timedelta(hours=over_threshold)
+        export.save()
+        pk = export.pk
+        delete_expired_failed_exports()
+        self.assertEquals(Export.objects.filter(pk=pk).first(), None)

onadata/settings/common.py

@@ -435,6 +435,7 @@ def configure_logging(logger, **kwargs):
 
 # number of records on export or CSV import before a progress update
 EXPORT_TASK_PROGRESS_UPDATE_BATCH = 1000
+EXPORT_TASK_LIFESPAN = 6  # six hours
 
 # default content length for submission requests
 DEFAULT_CONTENT_LENGTH = 10000000