mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-22 16:31:15 -05:00
754c65a6a2
make a decorator that pings cronitor before and after each task run. Designed for use with nightly tasks, so we have visibility if they fail. We have a bunch of cronitor monitors set up - 5 character keys that go into a URL that we then make a GET to with a self-explanatory url path (run/fail/complete). the cronitor URLs are defined in the credentials repo as a dictionary of celery task names to URL slugs. If the name passed in to the decorator isn't in that dict, it won't run. to use it, all you need to do is call `@cronitor(my_task_name)` instead of `@notify_celery.task`, and make sure that the task name and the matching slug are included in the credentials repo (or locally, json dumped and stored in the CRONITOR_KEYS environment variable)
343 lines
13 KiB
Python
343 lines
13 KiB
Python
from datetime import (
|
|
datetime,
|
|
timedelta
|
|
)
|
|
|
|
import pytz
|
|
from flask import current_app
|
|
from notifications_utils.statsd_decorators import statsd
|
|
from sqlalchemy import func
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from app import notify_celery, performance_platform_client, zendesk_client
|
|
from app.aws import s3
|
|
from app.celery.service_callback_tasks import (
|
|
send_delivery_status_to_service,
|
|
create_delivery_status_callback_data,
|
|
)
|
|
from app.config import QueueNames
|
|
from app.dao.inbound_sms_dao import delete_inbound_sms_created_more_than_a_week_ago
|
|
from app.dao.jobs_dao import (
|
|
dao_get_jobs_older_than_data_retention,
|
|
dao_archive_job
|
|
)
|
|
from app.dao.notifications_dao import (
|
|
dao_timeout_notifications,
|
|
delete_notifications_created_more_than_a_week_ago_by_type,
|
|
)
|
|
from app.dao.service_callback_api_dao import get_service_delivery_status_callback_api_for_service
|
|
from app.exceptions import NotificationTechnicalFailureException
|
|
from app.models import (
|
|
Notification,
|
|
NOTIFICATION_SENDING,
|
|
EMAIL_TYPE,
|
|
SMS_TYPE,
|
|
LETTER_TYPE,
|
|
KEY_TYPE_NORMAL
|
|
)
|
|
from app.performance_platform import total_sent_notifications, processing_time
|
|
from app.cronitor import cronitor
|
|
|
|
|
|
@notify_celery.task(name="remove_sms_email_jobs")
|
|
@cronitor("remove_sms_email_jobs")
|
|
@statsd(namespace="tasks")
|
|
def remove_sms_email_csv_files(job_types):
|
|
_remove_csv_files([EMAIL_TYPE, SMS_TYPE])
|
|
|
|
|
|
@notify_celery.task(name="remove_letter_jobs")
|
|
@cronitor("remove_letter_jobs")
|
|
@statsd(namespace="tasks")
|
|
def remove_letter_csv_files(job_types):
|
|
_remove_csv_files([LETTER_TYPE])
|
|
|
|
|
|
def _remove_csv_files(job_types):
|
|
jobs = dao_get_jobs_older_than_data_retention(notification_types=job_types)
|
|
for job in jobs:
|
|
s3.remove_job_from_s3(job.service_id, job.id)
|
|
dao_archive_job(job)
|
|
current_app.logger.info("Job ID {} has been removed from s3.".format(job.id))
|
|
|
|
|
|
@notify_celery.task(name="delete-sms-notifications")
|
|
@cronitor("delete-sms-notifications")
|
|
@statsd(namespace="tasks")
|
|
def delete_sms_notifications_older_than_seven_days():
|
|
try:
|
|
start = datetime.utcnow()
|
|
deleted = delete_notifications_created_more_than_a_week_ago_by_type('sms')
|
|
current_app.logger.info(
|
|
"Delete {} job started {} finished {} deleted {} sms notifications".format(
|
|
'sms',
|
|
start,
|
|
datetime.utcnow(),
|
|
deleted
|
|
)
|
|
)
|
|
except SQLAlchemyError:
|
|
current_app.logger.exception("Failed to delete sms notifications")
|
|
raise
|
|
|
|
|
|
@notify_celery.task(name="delete-email-notifications")
|
|
@cronitor("delete-email-notifications")
|
|
@statsd(namespace="tasks")
|
|
def delete_email_notifications_older_than_seven_days():
|
|
try:
|
|
start = datetime.utcnow()
|
|
deleted = delete_notifications_created_more_than_a_week_ago_by_type('email')
|
|
current_app.logger.info(
|
|
"Delete {} job started {} finished {} deleted {} email notifications".format(
|
|
'email',
|
|
start,
|
|
datetime.utcnow(),
|
|
deleted
|
|
)
|
|
)
|
|
except SQLAlchemyError:
|
|
current_app.logger.exception("Failed to delete email notifications")
|
|
raise
|
|
|
|
|
|
@notify_celery.task(name="delete-letter-notifications")
|
|
@cronitor("delete-letter-notifications")
|
|
@statsd(namespace="tasks")
|
|
def delete_letter_notifications_older_than_seven_days():
|
|
try:
|
|
start = datetime.utcnow()
|
|
deleted = delete_notifications_created_more_than_a_week_ago_by_type('letter')
|
|
current_app.logger.info(
|
|
"Delete {} job started {} finished {} deleted {} letter notifications".format(
|
|
'letter',
|
|
start,
|
|
datetime.utcnow(),
|
|
deleted
|
|
)
|
|
)
|
|
except SQLAlchemyError:
|
|
current_app.logger.exception("Failed to delete letter notifications")
|
|
raise
|
|
|
|
|
|
@notify_celery.task(name='timeout-sending-notifications')
|
|
@cronitor('timeout-sending-notifications')
|
|
@statsd(namespace="tasks")
|
|
def timeout_notifications():
|
|
technical_failure_notifications, temporary_failure_notifications = \
|
|
dao_timeout_notifications(current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD'))
|
|
|
|
notifications = technical_failure_notifications + temporary_failure_notifications
|
|
for notification in notifications:
|
|
# queue callback task only if the service_callback_api exists
|
|
service_callback_api = get_service_delivery_status_callback_api_for_service(service_id=notification.service_id)
|
|
if service_callback_api:
|
|
encrypted_notification = create_delivery_status_callback_data(notification, service_callback_api)
|
|
send_delivery_status_to_service.apply_async([str(notification.id), encrypted_notification],
|
|
queue=QueueNames.CALLBACKS)
|
|
|
|
current_app.logger.info(
|
|
"Timeout period reached for {} notifications, status has been updated.".format(len(notifications)))
|
|
if technical_failure_notifications:
|
|
message = "{} notifications have been updated to technical-failure because they " \
|
|
"have timed out and are still in created.Notification ids: {}".format(
|
|
len(technical_failure_notifications), [str(x.id) for x in technical_failure_notifications])
|
|
raise NotificationTechnicalFailureException(message)
|
|
|
|
|
|
@notify_celery.task(name='send-daily-performance-platform-stats')
|
|
@cronitor('send-daily-performance-platform-stats')
|
|
@statsd(namespace="tasks")
|
|
def send_daily_performance_platform_stats():
|
|
if performance_platform_client.active:
|
|
yesterday = datetime.utcnow() - timedelta(days=1)
|
|
send_total_sent_notifications_to_performance_platform(yesterday)
|
|
processing_time.send_processing_time_to_performance_platform()
|
|
|
|
|
|
def send_total_sent_notifications_to_performance_platform(day):
|
|
count_dict = total_sent_notifications.get_total_sent_notifications_for_day(day)
|
|
email_sent_count = count_dict.get('email').get('count')
|
|
sms_sent_count = count_dict.get('sms').get('count')
|
|
letter_sent_count = count_dict.get('letter').get('count')
|
|
start_date = count_dict.get('start_date')
|
|
|
|
current_app.logger.info(
|
|
"Attempting to update Performance Platform for {} with {} emails, {} text messages and {} letters"
|
|
.format(start_date, email_sent_count, sms_sent_count, letter_sent_count)
|
|
)
|
|
|
|
total_sent_notifications.send_total_notifications_sent_for_day_stats(
|
|
start_date,
|
|
'sms',
|
|
sms_sent_count
|
|
)
|
|
|
|
total_sent_notifications.send_total_notifications_sent_for_day_stats(
|
|
start_date,
|
|
'email',
|
|
email_sent_count
|
|
)
|
|
|
|
total_sent_notifications.send_total_notifications_sent_for_day_stats(
|
|
start_date,
|
|
'letter',
|
|
letter_sent_count
|
|
)
|
|
|
|
|
|
@notify_celery.task(name="delete-inbound-sms")
|
|
@cronitor("delete-inbound-sms")
|
|
@statsd(namespace="tasks")
|
|
def delete_inbound_sms_older_than_seven_days():
|
|
try:
|
|
start = datetime.utcnow()
|
|
deleted = delete_inbound_sms_created_more_than_a_week_ago()
|
|
current_app.logger.info(
|
|
"Delete inbound sms job started {} finished {} deleted {} inbound sms notifications".format(
|
|
start,
|
|
datetime.utcnow(),
|
|
deleted
|
|
)
|
|
)
|
|
except SQLAlchemyError:
|
|
current_app.logger.exception("Failed to delete inbound sms notifications")
|
|
raise
|
|
|
|
|
|
@notify_celery.task(name="remove_transformed_dvla_files")
|
|
@cronitor("remove_transformed_dvla_files")
|
|
@statsd(namespace="tasks")
|
|
def remove_transformed_dvla_files():
|
|
jobs = dao_get_jobs_older_than_data_retention(notification_types=[LETTER_TYPE])
|
|
for job in jobs:
|
|
s3.remove_transformed_dvla_file(job.id)
|
|
current_app.logger.info("Transformed dvla file for job {} has been removed from s3.".format(job.id))
|
|
|
|
|
|
# TODO: remove me, i'm not being run by anything
|
|
@notify_celery.task(name="delete_dvla_response_files")
|
|
@statsd(namespace="tasks")
|
|
def delete_dvla_response_files_older_than_seven_days():
|
|
try:
|
|
start = datetime.utcnow()
|
|
bucket_objects = s3.get_s3_bucket_objects(
|
|
current_app.config['DVLA_RESPONSE_BUCKET_NAME'],
|
|
'root/dispatch'
|
|
)
|
|
older_than_seven_days = s3.filter_s3_bucket_objects_within_date_range(bucket_objects)
|
|
|
|
for f in older_than_seven_days:
|
|
s3.remove_s3_object(current_app.config['DVLA_RESPONSE_BUCKET_NAME'], f['Key'])
|
|
|
|
current_app.logger.info(
|
|
"Delete dvla response files started {} finished {} deleted {} files".format(
|
|
start,
|
|
datetime.utcnow(),
|
|
len(older_than_seven_days)
|
|
)
|
|
)
|
|
except SQLAlchemyError:
|
|
current_app.logger.exception("Failed to delete dvla response files")
|
|
raise
|
|
|
|
|
|
@notify_celery.task(name="raise-alert-if-letter-notifications-still-sending")
|
|
@cronitor("raise-alert-if-letter-notifications-still-sending")
|
|
@statsd(namespace="tasks")
|
|
def raise_alert_if_letter_notifications_still_sending():
|
|
today = datetime.utcnow().date()
|
|
|
|
# Do nothing on the weekend
|
|
if today.isoweekday() in [6, 7]:
|
|
return
|
|
|
|
if today.isoweekday() in [1, 2]:
|
|
offset_days = 4
|
|
else:
|
|
offset_days = 2
|
|
still_sending = Notification.query.filter(
|
|
Notification.notification_type == LETTER_TYPE,
|
|
Notification.status == NOTIFICATION_SENDING,
|
|
Notification.key_type == KEY_TYPE_NORMAL,
|
|
func.date(Notification.sent_at) <= today - timedelta(days=offset_days)
|
|
).count()
|
|
|
|
if still_sending:
|
|
message = "There are {} letters in the 'sending' state from {}".format(
|
|
still_sending,
|
|
(today - timedelta(days=offset_days)).strftime('%A %d %B')
|
|
)
|
|
# Only send alerts in production
|
|
if current_app.config['NOTIFY_ENVIRONMENT'] in ['live', 'production', 'test']:
|
|
zendesk_client.create_ticket(
|
|
subject="[{}] Letters still sending".format(current_app.config['NOTIFY_ENVIRONMENT']),
|
|
message=message,
|
|
ticket_type=zendesk_client.TYPE_INCIDENT
|
|
)
|
|
else:
|
|
current_app.logger.info(message)
|
|
|
|
|
|
@notify_celery.task(name='raise-alert-if-no-letter-ack-file')
|
|
@cronitor('raise-alert-if-no-letter-ack-file')
|
|
@statsd(namespace="tasks")
|
|
def letter_raise_alert_if_no_ack_file_for_zip():
|
|
# get a list of zip files since yesterday
|
|
zip_file_set = set()
|
|
|
|
for key in s3.get_list_of_files_by_suffix(bucket_name=current_app.config['LETTERS_PDF_BUCKET_NAME'],
|
|
subfolder=datetime.utcnow().strftime('%Y-%m-%d') + '/zips_sent',
|
|
suffix='.TXT'):
|
|
subname = key.split('/')[-1] # strip subfolder in name
|
|
zip_file_set.add(subname.upper().rstrip('.TXT'))
|
|
|
|
# get acknowledgement file
|
|
ack_file_set = set()
|
|
|
|
yesterday = datetime.now(tz=pytz.utc) - timedelta(days=1) # AWS datetime format
|
|
|
|
for key in s3.get_list_of_files_by_suffix(bucket_name=current_app.config['DVLA_RESPONSE_BUCKET_NAME'],
|
|
subfolder='root/dispatch', suffix='.ACK.txt', last_modified=yesterday):
|
|
ack_file_set.add(key)
|
|
|
|
today_str = datetime.utcnow().strftime('%Y%m%d')
|
|
|
|
ack_content_set = set()
|
|
for key in ack_file_set:
|
|
if today_str in key:
|
|
content = s3.get_s3_file(current_app.config['DVLA_RESPONSE_BUCKET_NAME'], key)
|
|
for zip_file in content.split('\n'): # each line
|
|
s = zip_file.split('|')
|
|
ack_content_set.add(s[0].upper())
|
|
|
|
message = (
|
|
"Letter ack file does not contain all zip files sent. "
|
|
"Missing ack for zip files: {}, "
|
|
"pdf bucket: {}, subfolder: {}, "
|
|
"ack bucket: {}"
|
|
).format(
|
|
str(sorted(zip_file_set - ack_content_set)),
|
|
current_app.config['LETTERS_PDF_BUCKET_NAME'],
|
|
datetime.utcnow().strftime('%Y-%m-%d') + '/zips_sent',
|
|
current_app.config['DVLA_RESPONSE_BUCKET_NAME']
|
|
)
|
|
# strip empty element before comparison
|
|
ack_content_set.discard('')
|
|
zip_file_set.discard('')
|
|
|
|
if len(zip_file_set - ack_content_set) > 0:
|
|
if current_app.config['NOTIFY_ENVIRONMENT'] in ['live', 'production', 'test']:
|
|
zendesk_client.create_ticket(
|
|
subject="Letter acknowledge error",
|
|
message=message,
|
|
ticket_type=zendesk_client.TYPE_INCIDENT
|
|
)
|
|
current_app.logger.error(message)
|
|
|
|
if len(ack_content_set - zip_file_set) > 0:
|
|
current_app.logger.info(
|
|
"letter ack contains zip that is not for today: {}".format(ack_content_set - zip_file_set)
|
|
)
|