Merge pull request #2593 from alphagov/split-up-nightly-tasks

split up reporting tasks in to separate tasks per day

app/celery/reporting_tasks.py

@@ -12,11 +12,6 @@ from app.dao.fact_billing_dao import (
     update_fact_billing
 )
 from app.dao.fact_notification_status_dao import fetch_notification_status_for_day, update_fact_notification_status
-from app.celery.nightly_tasks import (
-    delete_sms_notifications_older_than_retention,
-    delete_email_notifications_older_than_retention,
-    delete_letter_notifications_older_than_retention
-)
 
 
 @notify_celery.task(name="create-nightly-billing")
@@ -33,13 +28,35 @@ def create_nightly_billing(day_start=None):
     for i in range(0, 4):
         process_day = day_start - timedelta(days=i)
 
-        transit_data = fetch_billing_data_for_day(process_day=process_day)
+        create_nightly_billing_for_day.apply_async(
+            kwargs={'process_day': process_day.isoformat()},
+            queue=QueueNames.REPORTING
+        )
 
-        for data in transit_data:
-            update_fact_billing(data, process_day)
 
-        current_app.logger.info(
-            "create-nightly-billing task complete. {} rows updated for day: {}".format(len(transit_data), process_day))
+@notify_celery.task(name="create-nightly-billing-for-day")
+@statsd(namespace="tasks")
+def create_nightly_billing_for_day(process_day):
+    process_day = datetime.strptime(process_day, "%Y-%m-%d").date()
+
+    start = datetime.utcnow()
+    transit_data = fetch_billing_data_for_day(process_day=process_day)
+    end = datetime.utcnow()
+
+    current_app.logger.info('create-nightly-billing-for-day {} fetched in {} seconds'.format(
+        process_day,
+        (end - start).seconds)
+    )
+
+    for data in transit_data:
+        update_fact_billing(data, process_day)
+
+    current_app.logger.info(
+        "create-nightly-billing-for-day task complete. {} rows updated for day: {}".format(
+            len(transit_data),
+            process_day
+        )
+    )
 
 
 @notify_celery.task(name="create-nightly-notification-status")
@@ -56,17 +73,29 @@ def create_nightly_notification_status(day_start=None):
     for i in range(0, 4):
         process_day = day_start - timedelta(days=i)
 
-        transit_data = fetch_notification_status_for_day(process_day=process_day)
-
-        update_fact_notification_status(transit_data, process_day)
-
-        current_app.logger.info(
-            "create-nightly-notification-status task: {} rows updated for day: {}".format(
-                len(transit_data), process_day
-            )
+        create_nightly_notification_status_for_day.apply_async(
+            kwargs={'process_day': process_day.isoformat()},
+            queue=QueueNames.REPORTING
         )
 
-    # delete jobs need to happen after nightly notification status is recorded to avoid conflict between the two tasks
-    delete_email_notifications_older_than_retention.apply_async(queue=QueueNames.PERIODIC)
-    delete_sms_notifications_older_than_retention.apply_async(queue=QueueNames.PERIODIC)
-    delete_letter_notifications_older_than_retention.apply_async(queue=QueueNames.PERIODIC)
+
+@notify_celery.task(name="create-nightly-notification-status-for-day")
+@statsd(namespace="tasks")
+def create_nightly_notification_status_for_day(process_day):
+    process_day = datetime.strptime(process_day, "%Y-%m-%d").date()
+
+    start = datetime.utcnow()
+    transit_data = fetch_notification_status_for_day(process_day=process_day)
+    end = datetime.utcnow()
+    current_app.logger.info('create-nightly-notification-status-for-day {} fetched in {} seconds'.format(
+        process_day,
+        (end - start).seconds)
+    )
+
+    update_fact_notification_status(transit_data, process_day)
+
+    current_app.logger.info(
+        "create-nightly-notification-status-for-day task complete: {} rows updated for day: {}".format(
+            len(transit_data), process_day
+        )
+    )

app/config.py

@@ -222,6 +222,21 @@ class Config(object):
             'schedule': crontab(hour=0, minute=30),  # after 'timeout-sending-notifications'
             'options': {'queue': QueueNames.REPORTING}
         },
+        'delete-sms-notifications': {
+            'task': 'delete-sms-notifications',
+            'schedule': crontab(hour=4, minute=15),  # after 'create-nightly-notification-status'
+            'options': {'queue': QueueNames.PERIODIC}
+        },
+        'delete-email-notifications': {
+            'task': 'delete-email-notifications',
+            'schedule': crontab(hour=4, minute=30),  # after 'create-nightly-notification-status'
+            'options': {'queue': QueueNames.PERIODIC}
+        },
+        'delete-letter-notifications': {
+            'task': 'delete-letter-notifications',
+            'schedule': crontab(hour=4, minute=45),  # after 'create-nightly-notification-status'
+            'options': {'queue': QueueNames.PERIODIC}
+        },
         'delete-inbound-sms': {
             'task': 'delete-inbound-sms',
             'schedule': crontab(hour=1, minute=40),

app/dao/fact_notification_status_dao.py

@@ -80,7 +80,8 @@ def query_for_fact_status_data(table, start_date, end_date, notification_type, s
         table.created_at >= start_date,
         table.created_at < end_date,
         table.notification_type == notification_type,
-        table.service_id == service_id
+        table.service_id == service_id,
+        table.key_type != KEY_TYPE_TEST
     ).group_by(
         table.template_id,
         table.service_id,