Merge pull request #3437 from alphagov/retry-parallel-status-180693991

Attempt 2 of parallelising status aggregation more

app/celery/nightly_tasks.py

@@ -22,7 +22,7 @@ from app.dao.jobs_dao import (
 from app.dao.notifications_dao import (
     dao_get_notifications_processing_time_stats,
     dao_timeout_notifications,
-    get_service_ids_that_have_notifications_from_before_timestamp,
+    get_service_ids_with_notifications_before,
     move_notifications_to_notification_history,
 )
 from app.dao.service_data_retention_dao import (
@@ -107,7 +107,7 @@ def _delete_notifications_older_than_retention_by_type(notification_type):
 
     # get a list of all service ids that we'll need to delete for. Typically that might only be 5% of services.
     # This query takes a couple of mins to run.
-    service_ids_that_have_sent_notifications_recently = get_service_ids_that_have_notifications_from_before_timestamp(
+    service_ids_that_have_sent_notifications_recently = get_service_ids_with_notifications_before(
         notification_type,
         seven_days_ago
     )

app/celery/reporting_tasks.py

@@ -14,7 +14,8 @@ from app.dao.fact_notification_status_dao import (
     fetch_status_data_for_service_and_day,
     update_fact_notification_status,
 )
-from app.models import EMAIL_TYPE, LETTER_TYPE, SMS_TYPE, Service
+from app.dao.notifications_dao import get_service_ids_with_notifications_on_date
+from app.models import EMAIL_TYPE, LETTER_TYPE, SMS_TYPE
 
 
 @notify_celery.task(name="create-nightly-billing")
@@ -97,52 +98,50 @@ def create_nightly_notification_status():
         for i in range(days):
             process_day = yesterday - timedelta(days=i)
 
-            create_nightly_notification_status_for_day.apply_async(
+            relevant_service_ids = get_service_ids_with_notifications_on_date(
-                kwargs={
+                notification_type, process_day
-                    'process_day': process_day.isoformat(),
-                    'notification_type': notification_type,
-                },
-                queue=QueueNames.REPORTING
-            )
-            current_app.logger.info(
-                f"create-nightly-notification-status-for-day task created "
-                f"for {notification_type} and {process_day}"
             )
 
+            for service_id in relevant_service_ids:
+                create_nightly_notification_status_for_service_and_day.apply_async(
+                    kwargs={
+                        'process_day': process_day.isoformat(),
+                        'notification_type': notification_type,
+                        'service_id': service_id,
+                    },
+                    queue=QueueNames.REPORTING
+                )
 
-@notify_celery.task(name="create-nightly-notification-status-for-day")
+
-def create_nightly_notification_status_for_day(process_day, notification_type):
+@notify_celery.task(name="create-nightly-notification-status-for-service-and-day")
+def create_nightly_notification_status_for_service_and_day(process_day, service_id, notification_type):
     process_day = datetime.strptime(process_day, "%Y-%m-%d").date()
-    current_app.logger.info(
-        f'create-nightly-notification-status-for-day task started '
-        f'for {notification_type} for {process_day}'
-    )
 
     start = datetime.utcnow()
-    new_status_rows = []
+    new_status_rows = fetch_status_data_for_service_and_day(
-
+        process_day=process_day,
-    for service in Service.query.all():
+        notification_type=notification_type,
-        new_status_rows += fetch_status_data_for_service_and_day(
+        service_id=service_id,
-            process_day=process_day,
+    )
-            notification_type=notification_type,
-            service_id=service.id,
-        )
 
     end = datetime.utcnow()
     current_app.logger.info(
         f'create-nightly-notification-status-for-day task fetch '
-        f'for {notification_type} for {process_day}: '
+        f'for {service_id}, {notification_type} for {process_day}: '
         f'data fetched in {(end - start).seconds} seconds'
     )
 
+    start = datetime.utcnow()
     update_fact_notification_status(
         new_status_rows=new_status_rows,
         process_day=process_day,
         notification_type=notification_type,
+        service_id=service_id
     )
 
+    end = datetime.utcnow()
     current_app.logger.info(
-        f'create-nightly-notification-status-for-day task finished '
+        f'create-nightly-notification-status-for-day task update '
-        f'for {notification_type} for {process_day}: '
+        f'for {service_id}, {notification_type} for {process_day}: '
-        f'{len(new_status_rows)} rows updated'
+        f'data fetched in {(end - start).seconds} seconds'
     )

app/dao/fact_billing_dao.py

@@ -1,7 +1,7 @@
-from datetime import date, datetime, time, timedelta
+from datetime import date, datetime, timedelta
 
 from flask import current_app
-from notifications_utils.timezones import convert_bst_to_utc, convert_utc_to_bst
+from notifications_utils.timezones import convert_utc_to_bst
 from sqlalchemy import Date, Integer, and_, desc, func
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.sql.expression import case, literal
@@ -317,8 +317,8 @@ def delete_billing_data_for_service_for_day(process_day, service_id):
 
 
 def fetch_billing_data_for_day(process_day, service_id=None, check_permissions=False):
-    start_date = convert_bst_to_utc(datetime.combine(process_day, time.min))
+    start_date = get_london_midnight_in_utc(process_day)
-    end_date = convert_bst_to_utc(datetime.combine(process_day + timedelta(days=1), time.min))
+    end_date = get_london_midnight_in_utc(process_day + timedelta(days=1))
     current_app.logger.info("Populate ft_billing for {} to {}".format(start_date, end_date))
     transit_data = []
     if not service_id:

app/dao/fact_notification_status_dao.py

@@ -1,6 +1,5 @@
-from datetime import datetime, time, timedelta
+from datetime import datetime, timedelta
 
-from notifications_utils.timezones import convert_bst_to_utc
 from sqlalchemy import Date, case, func
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.sql.expression import extract, literal
@@ -36,8 +35,8 @@ from app.utils import (
 
 
 def fetch_status_data_for_service_and_day(process_day, service_id, notification_type):
-    start_date = convert_bst_to_utc(datetime.combine(process_day, time.min))
+    start_date = get_london_midnight_in_utc(process_day)
-    end_date = convert_bst_to_utc(datetime.combine(process_day + timedelta(days=1), time.min))
+    end_date = get_london_midnight_in_utc(process_day + timedelta(days=1))
 
     # query notifications or notification_history for the day, depending on their data retention
     service = Service.query.get(service_id)
@@ -45,7 +44,6 @@ def fetch_status_data_for_service_and_day(process_day, service_id, notification_
 
     return db.session.query(
         table.template_id,
-        table.service_id,
         func.coalesce(table.job_id, '00000000-0000-0000-0000-000000000000').label('job_id'),
         table.key_type,
         table.status,
@@ -58,7 +56,6 @@ def fetch_status_data_for_service_and_day(process_day, service_id, notification_
         table.key_type.in_((KEY_TYPE_NORMAL, KEY_TYPE_TEAM)),
     ).group_by(
         table.template_id,
-        table.service_id,
         'job_id',
         table.key_type,
         table.status
@@ -66,11 +63,12 @@ def fetch_status_data_for_service_and_day(process_day, service_id, notification_
 
 
 @autocommit
-def update_fact_notification_status(new_status_rows, process_day, notification_type):
+def update_fact_notification_status(new_status_rows, process_day, notification_type, service_id):
     table = FactNotificationStatus.__table__
     FactNotificationStatus.query.filter(
         FactNotificationStatus.bst_date == process_day,
         FactNotificationStatus.notification_type == notification_type,
+        FactNotificationStatus.service_id == service_id,
     ).delete()
 
     for row in new_status_rows:
@@ -78,7 +76,7 @@ def update_fact_notification_status(new_status_rows, process_day, notification_t
             insert(table).values(
                 bst_date=process_day,
                 template_id=row.template_id,
-                service_id=row.service_id,
+                service_id=service_id,
                 job_id=row.job_id,
                 notification_type=notification_type,
                 key_type=row.key_type,

app/dao/notifications_dao.py

@@ -46,7 +46,11 @@ from app.models import (
     NotificationHistory,
     ProviderDetails,
 )
-from app.utils import escape_special_characters, midnight_n_days_ago
+from app.utils import (
+    escape_special_characters,
+    get_london_midnight_in_utc,
+    midnight_n_days_ago,
+)
 
 
 def dao_get_last_date_template_was_used(template_id, service_id):
@@ -784,7 +788,7 @@ def _duplicate_update_warning(notification, status):
     )
 
 
-def get_service_ids_that_have_notifications_from_before_timestamp(notification_type, timestamp):
+def get_service_ids_with_notifications_before(notification_type, timestamp):
     return {
         row.service_id
         for row in db.session.query(
@@ -794,3 +798,20 @@ def get_service_ids_that_have_notifications_from_before_timestamp(notification_t
             Notification.created_at < timestamp
         ).distinct()
     }
+
+
+def get_service_ids_with_notifications_on_date(notification_type, date):
+    start_date = get_london_midnight_in_utc(date)
+    end_date = get_london_midnight_in_utc(date + timedelta(days=1))
+
+    return {
+        row.service_id
+        for row in db.session.query(
+            Notification.service_id
+        ).filter(
+            Notification.notification_type == notification_type,
+            # using >= + < is much more efficient than date(created_at)
+            Notification.created_at >= start_date,
+            Notification.created_at < end_date,
+        ).distinct()
+    }

app/utils.py

@@ -8,7 +8,7 @@ from notifications_utils.template import (
     LetterPrintTemplate,
     SMSMessageTemplate,
 )
-from notifications_utils.timezones import convert_utc_to_bst
+from notifications_utils.timezones import convert_bst_to_utc, convert_utc_to_bst
 from sqlalchemy import func
 
 DATETIME_FORMAT_NO_TIMEZONE = "%Y-%m-%d %H:%M:%S.%f"
@@ -64,9 +64,7 @@ def get_london_midnight_in_utc(date):
      :param date: the day to calculate the London midnight in UTC for
      :return: the datetime of London midnight in UTC, for example 2016-06-17 = 2016-06-16 23:00:00
     """
-    return local_timezone.localize(datetime.combine(date, datetime.min.time())).astimezone(
+    return convert_bst_to_utc(datetime.combine(date, datetime.min.time()))
-        pytz.UTC).replace(
-        tzinfo=None)
 
 
 def get_midnight_for_day_before(date):

tests/app/celery/test_reporting_tasks.py

@@ -1,4 +1,3 @@
-import itertools
 from datetime import date, datetime, time, timedelta
 from decimal import Decimal
 from uuid import UUID
@@ -10,7 +9,7 @@ from app.celery.reporting_tasks import (
     create_nightly_billing,
     create_nightly_billing_for_day,
     create_nightly_notification_status,
-    create_nightly_notification_status_for_day,
+    create_nightly_notification_status_for_service_and_day,
 )
 from app.config import QueueNames
 from app.dao.fact_billing_dao import get_rate
@@ -20,6 +19,7 @@ from app.models import (
     KEY_TYPE_TEAM,
     KEY_TYPE_TEST,
     LETTER_TYPE,
+    NOTIFICATION_TYPES,
     SMS_TYPE,
     FactBilling,
     FactNotificationStatus,
@@ -61,37 +61,58 @@ def test_create_nightly_billing_triggers_tasks_for_days(notify_api, mocker, day_
         assert mock_celery.apply_async.call_args_list[i][1]['kwargs'] == {'process_day': expected_kwargs[i]}
 
 
-@freeze_time('2019-08-01')
+@freeze_time('2019-08-01T00:30')
-def test_create_nightly_notification_status_triggers_tasks(notify_api, sample_service, mocker):
+def test_create_nightly_notification_status_triggers_tasks(
-    mock_celery = mocker.patch('app.celery.reporting_tasks.create_nightly_notification_status_for_day')
+    notify_api,
+    sample_service,
+    sample_template,
+    mocker,
+):
+    mock_celery = mocker.patch(
+        'app.celery.reporting_tasks.create_nightly_notification_status_for_service_and_day'
+    ).apply_async
+
+    create_notification(template=sample_template, created_at='2019-07-31')
     create_nightly_notification_status()
 
-    assert mock_celery.apply_async.call_count == (
+    mock_celery.assert_called_with(
-        (4 * 3)  # four days, three notification types
+        kwargs={
-        +
+            'service_id': sample_service.id,
-        6  # six more days of just letters
+            'process_day': '2019-07-31',
+            'notification_type': SMS_TYPE
+        },
+        queue=QueueNames.REPORTING
     )
 
-    for process_date, notification_type in itertools.product(
-        ['2019-07-31', '2019-07-30', '2019-07-29', '2019-07-28'],
-        [SMS_TYPE, EMAIL_TYPE, LETTER_TYPE]
-    ):
-        mock_celery.apply_async.assert_any_call(
-            kwargs={
-                'process_day': process_date,
-                'notification_type': notification_type,
-            },
-            queue=QueueNames.REPORTING
-        )
 
-    for process_date in ['2019-07-27', '2019-07-26', '2019-07-25', '2019-07-24', '2019-07-23', '2019-07-22']:
+@freeze_time('2019-08-01T00:30')
-        mock_celery.apply_async.assert_any_call(
+@pytest.mark.parametrize('notification_date, expected_types_aggregated', [
-            kwargs={
+    ('2019-08-01', set()),
-                'process_day': process_date,
+    ('2019-07-31', {EMAIL_TYPE, SMS_TYPE, LETTER_TYPE}),
-                'notification_type': LETTER_TYPE,
+    ('2019-07-28', {EMAIL_TYPE, SMS_TYPE, LETTER_TYPE}),
-            },
+    ('2019-07-27', {LETTER_TYPE}),
-            queue=QueueNames.REPORTING
+    ('2019-07-22', {LETTER_TYPE}),
-        )
+    ('2019-07-21', set()),
+])
+def test_create_nightly_notification_status_triggers_relevant_tasks(
+    notify_api,
+    sample_service,
+    mocker,
+    notification_date,
+    expected_types_aggregated,
+):
+    mock_celery = mocker.patch(
+        'app.celery.reporting_tasks.create_nightly_notification_status_for_service_and_day'
+    ).apply_async
+
+    for notification_type in NOTIFICATION_TYPES:
+        template = create_template(sample_service, template_type=notification_type)
+        create_notification(template=template, created_at=notification_date)
+
+    create_nightly_notification_status()
+
+    types = {call.kwargs['kwargs']['notification_type'] for call in mock_celery.mock_calls}
+    assert types == expected_types_aggregated
 
 
 @pytest.mark.parametrize('second_rate, records_num, billable_units, multiplier',
@@ -507,7 +528,7 @@ def test_create_nightly_billing_for_day_update_when_record_exists(
     assert records[0].updated_at
 
 
-def test_create_nightly_notification_status_for_day(notify_db_session):
+def test_create_nightly_notification_status_for_service_and_day(notify_db_session):
     first_service = create_service(service_name='First Service')
     first_template = create_template(service=first_service)
     second_service = create_service(service_name='second Service')
@@ -538,9 +559,9 @@ def test_create_nightly_notification_status_for_day(notify_db_session):
 
     assert len(FactNotificationStatus.query.all()) == 0
 
-    create_nightly_notification_status_for_day(str(process_day), 'sms')
+    create_nightly_notification_status_for_service_and_day(str(process_day), first_service.id, 'sms')
-    create_nightly_notification_status_for_day(str(process_day), 'email')
+    create_nightly_notification_status_for_service_and_day(str(process_day), second_service.id, 'email')
-    create_nightly_notification_status_for_day(str(process_day), 'letter')
+    create_nightly_notification_status_for_service_and_day(str(process_day), third_service.id, 'letter')
 
     new_fact_data = FactNotificationStatus.query.order_by(
         FactNotificationStatus.notification_type
@@ -575,13 +596,13 @@ def test_create_nightly_notification_status_for_day(notify_db_session):
     assert new_fact_data[2].key_type == KEY_TYPE_NORMAL
 
 
-def test_create_nightly_notification_status_for_day_overwrites_old_data(notify_db_session):
+def test_create_nightly_notification_status_for_service_and_day_overwrites_old_data(notify_db_session):
     first_service = create_service(service_name='First Service')
     first_template = create_template(service=first_service)
     create_notification(template=first_template, status='delivered')
 
     process_day = date.today()
-    create_nightly_notification_status_for_day(str(process_day), 'sms')
+    create_nightly_notification_status_for_service_and_day(str(process_day), first_service.id, 'sms')
 
     new_fact_data = FactNotificationStatus.query.order_by(
         FactNotificationStatus.bst_date,
@@ -592,7 +613,7 @@ def test_create_nightly_notification_status_for_day_overwrites_old_data(notify_d
     assert new_fact_data[0].notification_count == 1
 
     create_notification(template=first_template, status='delivered')
-    create_nightly_notification_status_for_day(str(process_day), 'sms')
+    create_nightly_notification_status_for_service_and_day(str(process_day), first_service.id, 'sms')
 
     updated_fact_data = FactNotificationStatus.query.order_by(
         FactNotificationStatus.bst_date,
@@ -605,7 +626,7 @@ def test_create_nightly_notification_status_for_day_overwrites_old_data(notify_d
 
 # the job runs at 12:30am London time. 04/01 is in BST.
 @freeze_time('2019-04-01T23:30')
-def test_create_nightly_notification_status_for_day_respects_bst(sample_template):
+def test_create_nightly_notification_status_for_service_and_day_respects_bst(sample_template):
     create_notification(sample_template, status='delivered', created_at=datetime(2019, 4, 1, 23, 0))  # too new
 
     create_notification(sample_template, status='created', created_at=datetime(2019, 4, 1, 22, 59))
@@ -613,7 +634,7 @@ def test_create_nightly_notification_status_for_day_respects_bst(sample_template
 
     create_notification(sample_template, status='delivered', created_at=datetime(2019, 3, 31, 22, 59))  # too old
 
-    create_nightly_notification_status_for_day('2019-04-01', 'sms')
+    create_nightly_notification_status_for_service_and_day('2019-04-01', sample_template.service_id, 'sms')
 
     noti_status = FactNotificationStatus.query.order_by(FactNotificationStatus.bst_date).all()
     assert len(noti_status) == 1

tests/app/dao/notification_dao/test_notification_dao.py

@@ -1,5 +1,5 @@
 import uuid
-from datetime import datetime, timedelta
+from datetime import date, datetime, timedelta
 from functools import partial
 
 import pytest
@@ -24,6 +24,7 @@ from app.dao.notifications_dao import (
     get_notification_with_personalisation,
     get_notifications_for_job,
     get_notifications_for_service,
+    get_service_ids_with_notifications_on_date,
     is_delivery_slow_for_providers,
     notifications_not_yet_sent,
     update_notification_status_by_id,
@@ -41,6 +42,7 @@ from app.models import (
     NOTIFICATION_STATUS_TYPES,
     NOTIFICATION_STATUS_TYPES_FAILED,
     NOTIFICATION_TEMPORARY_FAILURE,
+    SMS_TYPE,
     Job,
     Notification,
     NotificationHistory,
@@ -1721,3 +1723,22 @@ def test_dao_get_letters_and_sheets_volume_by_postage(notify_db_session):
 
     for result in results:
         assert result._asdict() in expected_results
+
+
+@pytest.mark.parametrize('created_at_utc,date_to_check,expected_count', [
+    # Clocks change on the 27th of March 2022, so the query needs to look at the
+    # time range 00:00 - 23:00 (UTC) thereafter.
+    ('2022-03-27T00:30', date(2022, 3, 27), 1),  # 27/03 00:30 GMT
+    ('2022-03-27T22:30', date(2022, 3, 27), 1),  # 27/03 23:30 BST
+    ('2022-03-27T23:30', date(2022, 3, 27), 0),  # 28/03 00:30 BST
+    ('2022-03-26T23:30', date(2022, 3, 26), 1),  # 26/03 23:30 GMT
+])
+def test_get_service_ids_with_notifications_on_date_respects_gmt_bst(
+    sample_template,
+    created_at_utc,
+    date_to_check,
+    expected_count
+):
+    create_notification(template=sample_template, created_at=created_at_utc)
+    service_ids = get_service_ids_with_notifications_on_date(SMS_TYPE, date_to_check)
+    assert len(service_ids) == expected_count

tests/app/dao/test_fact_notification_status_dao.py

@@ -14,6 +14,7 @@ from app.dao.fact_notification_status_dao import (
     fetch_notification_status_totals_for_all_services,
     fetch_notification_statuses_for_job,
     fetch_stats_for_all_services_by_date_range,
+    fetch_status_data_for_service_and_day,
     get_total_notifications_for_date_range,
 )
 from app.models import (
@@ -607,3 +608,23 @@ def test_get_total_notifications_for_date_range(sample_service):
 
     assert len(results) == 1
     assert results[0] == ("2021-03-01", 15, 20, 3)
+
+
+@pytest.mark.parametrize('created_at_utc,process_day,expected_count', [
+    # Clocks change on the 27th of March 2022, so the query needs to look at the
+    # time range 00:00 - 23:00 (UTC) thereafter.
+    ('2022-03-27T00:30', date(2022, 3, 27), 1),  # 27/03 00:30 GMT
+    ('2022-03-27T22:30', date(2022, 3, 27), 1),  # 27/03 23:30 BST
+    ('2022-03-27T23:30', date(2022, 3, 27), 0),  # 28/03 00:30 BST
+    ('2022-03-26T23:30', date(2022, 3, 26), 1),  # 26/03 23:30 GMT
+])
+def test_fetch_status_data_for_service_and_day_respects_gmt_bst(
+    sample_template,
+    sample_service,
+    created_at_utc,
+    process_day,
+    expected_count,
+):
+    create_notification(template=sample_template, created_at=created_at_utc)
+    rows = fetch_status_data_for_service_and_day(process_day, sample_service.id, SMS_TYPE)
+    assert len(rows) == expected_count