From 33d85322c96800b3135de63aa3e3b045b87afca5 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Fri, 20 Mar 2020 14:45:00 +0000 Subject: [PATCH] Change sql to chunk by hour to remove old email notifications So since removing the subquery in the previous commit, we now are doing all of the inserts using offsets and limits to group by 10,000 and deleting all records in a single query (which could be up to as many as 10 million rows). We want to avoid doing this, because both of these ways we think are going to result in expensive queries. Therefore we have introduced our own chunking of the notifications by hour periods meaning we do not need to use offset and limits. We estimate that GOV.UK email will at most send 600,000 notifications per hour (175 per second * 60 * 60). --- app/dao/notifications_dao.py | 110 +++++++++++------- ...t_notification_dao_delete_notifications.py | 62 ++++------ tests/app/db.py | 5 +- 3 files changed, 91 insertions(+), 86 deletions(-) diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py index 0d6ba4839..3b313a0df 100644 --- a/app/dao/notifications_dao.py +++ b/app/dao/notifications_dao.py @@ -307,19 +307,14 @@ def delete_notifications_older_than_retention_by_type(notification_type, qry_lim ).all() deleted = 0 for f in flexible_data_retention: - days_of_retention = get_london_midnight_in_utc( - convert_utc_to_bst(datetime.utcnow()).date()) - timedelta(days=f.days_of_retention) - - if notification_type == LETTER_TYPE: - _delete_letters_from_s3( - notification_type, f.service_id, days_of_retention, qry_limit - ) - - insert_update_notification_history(notification_type, days_of_retention, f.service_id) - current_app.logger.info( "Deleting {} notifications for service id: {}".format(notification_type, f.service_id)) - deleted += _delete_notifications(notification_type, days_of_retention, f.service_id, qry_limit) + + day_to_delete_backwards_from = get_london_midnight_in_utc( + convert_utc_to_bst(datetime.utcnow()).date()) - timedelta(days=f.days_of_retention) + + deleted += _move_notifications_to_notification_history( + notification_type, f.service_id, day_to_delete_backwards_from, qry_limit) current_app.logger.info( 'Deleting {} notifications for services without flexible data retention'.format(notification_type)) @@ -329,31 +324,66 @@ def delete_notifications_older_than_retention_by_type(notification_type, qry_lim service_ids_to_purge = db.session.query(Service.id).filter(Service.id.notin_(services_with_data_retention)).all() for service_id in service_ids_to_purge: - if notification_type == LETTER_TYPE: - _delete_letters_from_s3( - notification_type, service_id, seven_days_ago, qry_limit - ) - insert_update_notification_history(notification_type, seven_days_ago, service_id) - deleted += _delete_notifications(notification_type, seven_days_ago, service_id, qry_limit) + deleted += _move_notifications_to_notification_history( + notification_type, service_id, seven_days_ago, qry_limit) current_app.logger.info('Finished deleting {} notifications'.format(notification_type)) return deleted -def _delete_notifications(notification_type, date_to_delete_from, service_id, query_limit): +def _move_notifications_to_notification_history(notification_type, service_id, day_to_delete_backwards_from, qry_limit): + deleted = 0 + if notification_type == LETTER_TYPE: + _delete_letters_from_s3( + notification_type, service_id, day_to_delete_backwards_from, qry_limit + ) + + stop = -1 # exclusive, we want to include 0 + step = -1 + for hour_delta in range(23, stop, step): + # We find the timestamp we want to delete all notifications backwards from + # We then start 23 hours ago, and do an insert notification history before deleting all notifications older + # We then look 22 hours ago, do an insert notifications history before deleting all notifications older + # We continue this until we reach the original timestamp we wanted to delete notifications backwardsfrom + # This enables us to break this into smaller database queries + timestamp_to_delete_backwards_from = day_to_delete_backwards_from - timedelta(hours=hour_delta) + + if service_id == '539d63a1-701d-400d-ab11-f3ee2319d4d4': + current_app.logger.info( + "Beginning insert_update_notification_history for GOV.UK Email from {} backwards".format( + timestamp_to_delete_backwards_from + ) + ) + + insert_update_notification_history(notification_type, timestamp_to_delete_backwards_from, service_id) + + if service_id == '539d63a1-701d-400d-ab11-f3ee2319d4d4': + current_app.logger.info( + "Beginning _delete_notifications for GOV.UK Email {} backwards".format( + timestamp_to_delete_backwards_from + ) + ) + + deleted += _delete_notifications( + notification_type, timestamp_to_delete_backwards_from, service_id + ) + + return deleted + + +def _delete_notifications(notification_type, date_to_delete_from, service_id): deleted = Notification.query.filter( Notification.notification_type == notification_type, Notification.service_id == service_id, Notification.created_at < date_to_delete_from, ).delete() + db.session.commit() return deleted -def insert_update_notification_history(notification_type, date_to_delete_from, service_id, query_limit=10000): - offset = 0 - +def insert_update_notification_history(notification_type, date_to_delete_from, service_id): notification_query = db.session.query( *[x.name for x in NotificationHistory.__table__.c] ).filter( @@ -362,29 +392,25 @@ def insert_update_notification_history(notification_type, date_to_delete_from, s Notification.created_at < date_to_delete_from, Notification.key_type != KEY_TYPE_TEST ) - notifications_count = notification_query.count() - while offset < notifications_count: - stmt = insert(NotificationHistory).from_select( - NotificationHistory.__table__.c, - notification_query.limit(query_limit).offset(offset) - ) + stmt = insert(NotificationHistory).from_select( + NotificationHistory.__table__.c, + notification_query + ) - stmt = stmt.on_conflict_do_update( - constraint="notification_history_pkey", - set_={ - "notification_status": stmt.excluded.status, - "reference": stmt.excluded.reference, - "billable_units": stmt.excluded.billable_units, - "updated_at": stmt.excluded.updated_at, - "sent_at": stmt.excluded.sent_at, - "sent_by": stmt.excluded.sent_by - } - ) - db.session.connection().execute(stmt) - db.session.commit() - - offset += query_limit + stmt = stmt.on_conflict_do_update( + constraint="notification_history_pkey", + set_={ + "notification_status": stmt.excluded.status, + "reference": stmt.excluded.reference, + "billable_units": stmt.excluded.billable_units, + "updated_at": stmt.excluded.updated_at, + "sent_at": stmt.excluded.sent_at, + "sent_by": stmt.excluded.sent_by + } + ) + db.session.connection().execute(stmt) + db.session.commit() def _delete_letters_from_s3( diff --git a/tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py b/tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py index 202019ed5..0869d2f6a 100644 --- a/tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py +++ b/tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py @@ -9,13 +9,13 @@ from freezegun import freeze_time from app.dao.notifications_dao import ( delete_notifications_older_than_retention_by_type, - db, insert_update_notification_history ) from app.models import Notification, NotificationHistory from tests.app.db import ( create_template, create_notification, + create_notification_history, create_service_data_retention, create_service ) @@ -75,7 +75,7 @@ def test_should_delete_notifications_by_type_after_seven_days( ): mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") email_template, letter_template, sms_template = _create_templates(sample_service) - # create one notification a day between 1st and 10th from 11:00 to 19:00 of each type + # create one notification a day between 1st and 10th from 01:00 to 11:00 of each type for i in range(1, 11): past_date = '2016-0{0}-{1:02d} {1:02d}:00:00.000000'.format(month, i) with freeze_time(past_date): @@ -157,17 +157,24 @@ def test_delete_notifications_inserts_notification_history(sample_service): assert NotificationHistory.query.count() == 2 -def test_delete_notifications_updates_notification_history(sample_email_template, mocker): +def test_delete_notifications_updates_notification_history(notify_db, sample_email_template, mocker): mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects") - notification = create_notification(template=sample_email_template, created_at=datetime.utcnow() - timedelta(days=8)) - Notification.query.filter_by(id=notification.id).update( - {"status": "delivered", - "reference": "ses_reference", - "billable_units": 1, # I know we don't update this for emails but this is a unit test - "updated_at": datetime.utcnow(), - "sent_at": datetime.utcnow(), - "sent_by": "ses" - } + now = datetime.utcnow() + notification = create_notification( + template=sample_email_template, + created_at=datetime.utcnow() - timedelta(days=8), + reference="ses_reference", + billable_units=1, + updated_at=now, + sent_by="ses", + status='delivered' + ) + + create_notification_history( + id=notification.id, + template=sample_email_template, + created_at=datetime.utcnow() - timedelta(days=8), + status='sending', ) delete_notifications_older_than_retention_by_type("email") @@ -177,7 +184,7 @@ def test_delete_notifications_updates_notification_history(sample_email_template assert history[0].status == 'delivered' assert history[0].reference == 'ses_reference' assert history[0].billable_units == 1 - assert history[0].updated_at + assert history[0].updated_at == now assert history[0].sent_by == 'ses' @@ -273,35 +280,6 @@ def test_insert_update_notification_history(sample_service): assert notification_3.id in history_ids -def test_insert_update_notification_history_with_more_notifications_than_query_limit(mocker, sample_service): - template = create_template(sample_service, template_type='sms') - notification_1 = create_notification(template=template, created_at=datetime.utcnow() - timedelta(days=3)) - notification_2 = create_notification(template=template, created_at=datetime.utcnow() - timedelta(days=8)) - notification_3 = create_notification(template=template, created_at=datetime.utcnow() - timedelta(days=9)) - other_types = ['email', 'letter'] - for template_type in other_types: - t = create_template(service=sample_service, template_type=template_type) - create_notification(template=t, created_at=datetime.utcnow() - timedelta(days=3)) - create_notification(template=t, created_at=datetime.utcnow() - timedelta(days=8)) - - db_connection_spy = mocker.spy(db.session, 'connection') - db_commit_spy = mocker.spy(db.session, 'commit') - - insert_update_notification_history( - notification_type='sms', date_to_delete_from=datetime.utcnow() - timedelta(days=7), - service_id=sample_service.id, query_limit=1) - history = NotificationHistory.query.all() - assert len(history) == 2 - - history_ids = [x.id for x in history] - assert notification_1.id not in history_ids - assert notification_2.id in history_ids - assert notification_3.id in history_ids - - assert db_connection_spy.call_count == 2 - assert db_commit_spy.call_count == 2 - - def test_insert_update_notification_history_only_insert_update_given_service(sample_service): other_service = create_service(service_name='another service') other_template = create_template(service=other_service) diff --git a/tests/app/db.py b/tests/app/db.py index 1b870924b..f9ed21efe 100644 --- a/tests/app/db.py +++ b/tests/app/db.py @@ -324,7 +324,8 @@ def create_notification_history( international=False, phone_prefix=None, created_by_id=None, - postage=None + postage=None, + id=None ): assert job or template if job: @@ -341,7 +342,7 @@ def create_notification_history( postage = 'second' data = { - 'id': uuid.uuid4(), + 'id': id or uuid.uuid4(), 'job_id': job and job.id, 'job': job, 'service_id': template.service.id,