From 4adab1afeb45aab1789ff5e2433c4528b3a0a144 Mon Sep 17 00:00:00 2001 From: Rebecca Law Date: Fri, 20 Mar 2020 09:34:07 +0000 Subject: [PATCH] First draft to purge high volume data on an hourly basis. This makes the assumption that we do not need NotificationHistory, we will need to ensure that for HighVolumeServices we do not recalculate the stats tables, either in the reporting_tasks nor in the commands. The strategy to migrate to using this code is still up for design. We will need to consider: - how to purge the data, safely, efficiently. - ensure we get the counts right for the stats table, the ft_notification_status table is particularly tricky, because as a notification moves from created -> sending -> delivered the row counts change for each status type. One idea is that we only write the `delivered`, `temporary-failure` and `permanent-failure` status into the table. This is not production ready but just a draft to get the conversation going. I also have an idea to stop returning "todays" notification data for HighVolumeServices on the dashboards/usages (except platform_admin) until this code is ready. --- app/celery/reporting_tasks.py | 6 + app/celery/scheduled_tasks.py | 114 +++++++++++++++++- app/models.py | 7 ++ .../versions/0319_high_volume_service.py | 27 +++++ tests/app/celery/test_scheduled_tasks.py | 60 ++++++++- 5 files changed, 209 insertions(+), 5 deletions(-) create mode 100644 migrations/versions/0319_high_volume_service.py diff --git a/app/celery/reporting_tasks.py b/app/celery/reporting_tasks.py index 7ce1ff57a..ea5e50f2b 100644 --- a/app/celery/reporting_tasks.py +++ b/app/celery/reporting_tasks.py @@ -42,6 +42,8 @@ def create_nightly_billing(day_start=None): @notify_celery.task(name="create-nightly-billing-for-day") @statsd(namespace="tasks") def create_nightly_billing_for_day(process_day): + # When app.celery.scheduled_tasks.purge_high_volume_notifications starts + # we need to exclude HighVolumeServices from the list. process_day = datetime.strptime(process_day, "%Y-%m-%d").date() start = datetime.utcnow() @@ -62,6 +64,8 @@ def create_nightly_billing_for_day(process_day): @cronitor("create-nightly-notification-status") @statsd(namespace="tasks") def create_nightly_notification_status(): + # When app.celery.scheduled_tasks.purge_high_volume_notifications starts + # we need to exclude HighVolumeServices from the list. yesterday = convert_utc_to_bst(datetime.utcnow()).date() - timedelta(days=1) # email and sms @@ -84,6 +88,8 @@ def create_nightly_notification_status(): @notify_celery.task(name="create-nightly-notification-status-for-day") @statsd(namespace="tasks") def create_nightly_notification_status_for_day(process_day, notification_type): + # When app.celery.scheduled_tasks.purge_high_volume_notifications starts + # we need to exclude HighVolumeServices from the list. process_day = datetime.strptime(process_day, "%Y-%m-%d").date() start = datetime.utcnow() diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py index e0c419534..6c56f6eb1 100644 --- a/app/celery/scheduled_tasks.py +++ b/app/celery/scheduled_tasks.py @@ -5,10 +5,11 @@ from datetime import ( from flask import current_app from notifications_utils.statsd_decorators import statsd +from notifications_utils.timezones import convert_utc_to_bst from sqlalchemy import and_ from sqlalchemy.exc import SQLAlchemyError -from app import notify_celery, zendesk_client +from app import notify_celery, zendesk_client, db from app.celery.tasks import ( process_job, get_recipient_csv_and_template_and_sender_id, @@ -45,7 +46,7 @@ from app.models import ( JOB_STATUS_ERROR, SMS_TYPE, EMAIL_TYPE, -) + HighVolumeService, Template, Notification, KEY_TYPE_NORMAL, FactBilling, FactNotificationStatus) from app.notifications.process_notifications import send_notification_to_queue from app.v2.errors import JobIncompleteError @@ -311,3 +312,112 @@ def check_for_services_with_high_failure_rates_or_sending_to_tv_numbers(): message=message, ticket_type=zendesk_client.TYPE_INCIDENT ) + + +def purge_high_volume_notifications(): + # Not sure what time to use here..... + hour_ago = datetime.utcnow() - timedelta(hours=1) + + # We could also store the id in config... however this does give us a bit of control, + # as in we can stop this process from happening by deleting the row. + # The other part of it is to potentially take this tactic for other services... + # perhaps hour_ago is actually the timedelta from the table. + services = HighVolumeService.query.all() + + for service in services: + templates = Template.query.filter_by( + service_id=service.service_id, + # going to assume we are only dealing with emails because I don't want to deal with rates at this time. + template_type=EMAIL_TYPE + ).all() + for status in ['delivered', 'temporary-failure', 'permanent-failure']: + for template in templates: + del_count = Notification.query.filter( + Notification.service_id == service.service_id, + Notification.template_id == template.id, + Notification.notification_type == template.template_type, + Notification.status == status, + Notification.key_type == KEY_TYPE_NORMAL, + Notification.created_at < hour_ago, + ).delete() + + bst_date = convert_utc_to_bst(hour_ago).date() + # upsert stat data + upsert_ft_billing(bst_date, del_count, service.service_id, template) + upsert_ft_notification_status(bst_date, del_count, service.service_id, status, template) + + +def upsert_ft_notification_status(bst_date, del_count, service_id, status, template): + ft_status_row = FactNotificationStatus.query.filter( + FactNotificationStatus.service_id == service_id, + FactNotificationStatus.template_id == template.id, + FactNotificationStatus.bst_date == bst_date, + FactNotificationStatus.notification_type == template.template_type, + FactNotificationStatus.notification_status == status, + FactNotificationStatus.key_type == KEY_TYPE_NORMAL + ).first() + if ft_status_row: + # How do we deal with rows in ft_status where there are.... this isn't going to work. + # The current process take the days total from Notifications, + # deletes the row for status then inserts, + # this means that any rows with a created status should eventually go away. + # Do we stop processing the HighVolumes services in the reporting tasks? + # If yes, then we need to thing about the migration plan, take a snapshot of data first, etc. + FactNotificationStatus.query.filter( + FactNotificationStatus.service_id == service_id, + FactNotificationStatus.template_id == template.id, + FactNotificationStatus.bst_date == bst_date, + FactNotificationStatus.notification_type == template.template_type, + FactNotificationStatus.notification_status == status, + FactNotificationStatus.key_type == KEY_TYPE_NORMAL + ).update({"notification_count": ft_status_row.notification_count + del_count}) + else: + ft_status = FactNotificationStatus( + bst_date=bst_date, + template_id=template.id, + service_id=service_id, + job_id='00000000-0000-0000-0000-000000000000', + notification_type=template.template_type, + key_type=KEY_TYPE_NORMAL, + notification_status=status, + notification_count=del_count, + created_at=datetime.utcnow() + ) + db.session.add(ft_status) + db.session.commit() + + +def upsert_ft_billing(bst_date, del_count, service_id, template): + ft_billing_row = FactBilling.query.filter( + FactBilling.service_id == service_id, + FactBilling.template_id == template.id, + FactBilling.bst_date == bst_date, + FactBilling.notification_type == template.template_type + ).first() + if not ft_billing_row: + # insert new row + ft_billing = FactBilling( + bst_date=bst_date, + service_id=service_id, + template_id=template.id, + notification_type=template.template_type, + provider='SES', + rate_multiplier=0, + international=False, + rate=0, + billable_units=0, + notifications_sent=del_count, + created_at=datetime.utcnow(), + postage='none' + ) + db.session.add(ft_billing) + db.session.commit() + + else: + FactBilling.query.filter( + FactBilling.service_id == service_id, + FactBilling.template_id == template.id, + FactBilling.bst_date == bst_date, + FactBilling.notification_type == template.template_type + ).update({'notifications_sent': ft_billing_row.notifications_sent + del_count}) + db.session.commit() diff --git a/app/models.py b/app/models.py index 802567046..d1c3d5c1d 100644 --- a/app/models.py +++ b/app/models.py @@ -2148,3 +2148,10 @@ class ServiceContactList(db.Model): "created_at": created_at_in_bst.strftime("%Y-%m-%d %H:%M:%S"), } return contact_list + + +class HighVolumeService(db.Model): + # Service that we want to purge data for hourly + __tablename__ = 'high_volume_service' + + service_id = db.Column(UUID(as_uuid=True), primary_key=True, unique=True, index=True, nullable=False) diff --git a/migrations/versions/0319_high_volume_service.py b/migrations/versions/0319_high_volume_service.py new file mode 100644 index 000000000..4589d32b6 --- /dev/null +++ b/migrations/versions/0319_high_volume_service.py @@ -0,0 +1,27 @@ +""" + +Revision ID: 0319_high_volume_service +Revises: 0318_service_contact_list +Create Date: 2020-03-20 08:53:22.624516 + +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision = '0319_high_volume_service' +down_revision = '0318_service_contact_list' + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('high_volume_service', + sa.Column('service_id', postgresql.UUID(as_uuid=True), nullable=False), + sa.PrimaryKeyConstraint('service_id') + ) + op.create_index(op.f('ix_high_volume_service_service_id'), 'high_volume_service', ['service_id'], unique=True) + + +def downgrade(): + op.drop_index(op.f('ix_high_volume_service_service_id'), table_name='high_volume_service') + op.drop_table('high_volume_service') diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py index bb43251c6..e266b13a6 100644 --- a/tests/app/celery/test_scheduled_tasks.py +++ b/tests/app/celery/test_scheduled_tasks.py @@ -6,6 +6,7 @@ from collections import namedtuple from freezegun import freeze_time from mock import mock +from app import db from app.celery import scheduled_tasks from app.celery.scheduled_tasks import ( check_job_status, @@ -19,7 +20,7 @@ from app.celery.scheduled_tasks import ( check_for_missing_rows_in_completed_jobs, check_for_services_with_high_failure_rates_or_sending_to_tv_numbers, switch_current_sms_provider_on_slow_delivery, -) + purge_high_volume_notifications) from app.config import QueueNames, TaskNames, Config from app.dao.jobs_dao import dao_get_job_by_id from app.dao.notifications_dao import dao_get_scheduled_notifications @@ -30,7 +31,7 @@ from app.models import ( JOB_STATUS_FINISHED, NOTIFICATION_DELIVERED, NOTIFICATION_PENDING_VIRUS_CHECK, -) + HighVolumeService, Notification, FactBilling, FactNotificationStatus) from app.v2.errors import JobIncompleteError from tests.app import load_example_csv @@ -314,7 +315,7 @@ def test_replay_created_notifications(notify_db_session, sample_service, mocker) def test_replay_created_notifications_create_letters_pdf_tasks_for_letters_not_ready_to_send( - sample_letter_template, mocker + sample_letter_template, mocker ): mock_task = mocker.patch('app.celery.scheduled_tasks.create_letters_pdf.apply_async') create_notification(template=sample_letter_template, billable_units=0, @@ -556,3 +557,56 @@ def test_check_for_services_with_high_failure_rates_or_sending_to_tv_numbers( subject="[test] High failure rates for sms spotted for services", ticket_type='incident' ) + + +@freeze_time('2020-03-19 13:30') +def test_purge_high_volume_notifications(sample_email_template, notify_db_session): + # should be deleted + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(days=4), status='delivered') + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(hours=2), status='permanent-failure') + + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(hours=1, minutes=1), status='temporary-failure') + # should NOT be deleted + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(minutes=59), status='temporary-failure') + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(hours=1), status='temporary-failure') + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(hours=1), status='delivered') + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(hours=1), status='created') + + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(days=1), status='sending') + + create_notification(template=sample_email_template, + created_at=datetime.utcnow() - timedelta(days=1), status='technical-failure') + + high_volume_service = HighVolumeService(service_id=sample_email_template.service_id) + db.session.add(high_volume_service) + db.session.commit() + purge_high_volume_notifications() + + notifications = Notification.query.all() + assert len(notifications) == 6 + + ft_billing = FactBilling.query.all() + assert len(ft_billing) == 1 + assert ft_billing[0].service_id == sample_email_template.service_id + assert ft_billing[0].notifications_sent == 3 + assert str(ft_billing[0].bst_date) == '2020-03-19' + + ft_status = FactNotificationStatus.query.order_by(FactNotificationStatus.notification_status).all() + assert len(ft_status) == 3 + assert str(ft_status[0].bst_date) == '2020-03-19' + assert ft_status[0].notification_status == 'delivered' + assert ft_status[0].notification_count == 1 + assert str(ft_status[1].bst_date) == '2020-03-19' + assert ft_status[1].notification_status == 'permanent-failure' + assert ft_status[1].notification_count == 1 + assert str(ft_status[2].bst_date) == '2020-03-19' + assert ft_status[2].notification_status == 'temporary-failure' + assert ft_status[2].notification_count == 1