Merge branch 'main' of https://github.com/GSA/notifications-api into notify-233b

app/celery/nightly_tasks.py

@@ -4,8 +4,9 @@ from flask import current_app
 from notifications_utils.timezones import convert_utc_to_local_timezone
 from sqlalchemy.exc import SQLAlchemyError
 
-from app import notify_celery, statsd_client
+from app import notify_celery
 from app.aws import s3
+from app.aws.s3 import remove_csv_object
 from app.celery.process_ses_receipts_tasks import check_and_queue_callback_task
 from app.config import QueueNames
 from app.cronitor import cronitor
@@ -14,6 +15,7 @@ from app.dao.inbound_sms_dao import delete_inbound_sms_older_than_retention
 from app.dao.jobs_dao import (
     dao_archive_job,
     dao_get_jobs_older_than_data_retention,
+    dao_get_unfinished_jobs,
 )
 from app.dao.notifications_dao import (
     dao_get_notifications_processing_time_stats,
@@ -42,6 +44,19 @@ def _remove_csv_files(job_types):
         current_app.logger.info("Job ID {} has been removed from s3.".format(job.id))
 
 
+@notify_celery.task(name="cleanup-unfinished-jobs")
+def cleanup_unfinished_jobs():
+    now = datetime.utcnow()
+    jobs = dao_get_unfinished_jobs()
+    for job in jobs:
+        # The query already checks that the processing_finished time is null, so here we are saying
+        # if it started more than 4 hours ago, that's too long
+        acceptable_finish_time = job.processing_started + timedelta(minutes=5)
+        if now > acceptable_finish_time:
+            remove_csv_object(job.original_file_name)
+            dao_archive_job(job)
+
+
 @notify_celery.task(name="delete-notifications-older-than-retention")
 def delete_notifications_older_than_retention():
     delete_email_notifications_older_than_retention.apply_async(queue=QueueNames.REPORTING)
@@ -134,7 +149,6 @@ def timeout_notifications():
         notifications = dao_timeout_notifications(cutoff_time)
 
         for notification in notifications:
-            statsd_client.incr(f'timeout-sending.{notification.sent_by}')
             check_and_queue_callback_task(notification)
 
         current_app.logger.info(
@@ -162,6 +176,7 @@ def delete_inbound_sms():
 @notify_celery.task(name='save-daily-notification-processing-time')
 @cronitor("save-daily-notification-processing-time")
 def save_daily_notification_processing_time(local_date=None):
+
     # local_date is a string in the format of "YYYY-MM-DD"
     if local_date is None:
         # if a date is not provided, we run against yesterdays data

app/celery/process_ses_receipts_tasks.py

@@ -5,7 +5,7 @@ from celery.exceptions import Retry
 from flask import current_app, json
 from sqlalchemy.orm.exc import NoResultFound
 
-from app import notify_celery, statsd_client
+from app import notify_celery
 from app.celery.service_callback_tasks import (
     create_complaint_callback_data,
     create_delivery_status_callback_data,
@@ -92,11 +92,6 @@ def process_ses_results(self, response):
                 "SES callback return status of {} for notification: {}".format(notification_status, notification.id)
             )
 
-        statsd_client.incr("callback.ses.{}".format(notification_status))
-
-        if notification.sent_at:
-            statsd_client.timing_with_dates("callback.ses.elapsed-time", datetime.utcnow(), notification.sent_at)
-
         check_and_queue_callback_task(notification)
 
         return True

app/celery/provider_tasks.py

@@ -1,7 +1,11 @@
+from datetime import datetime, timedelta
+from time import time
+from zoneinfo import ZoneInfo
+
 from flask import current_app
 from sqlalchemy.orm.exc import NoResultFound
 
-from app import notify_celery
+from app import aws_cloudwatch_client, notify_celery
 from app.clients.email import EmailClientNonRetryableException
 from app.clients.email.aws_ses import AwsSesClientThrottlingSendRateException
 from app.clients.sms import SmsClientResponseException
@@ -13,17 +17,51 @@ from app.dao.notifications_dao import (
 )
 from app.delivery import send_to_providers
 from app.exceptions import NotificationTechnicalFailureException
-from app.models import NOTIFICATION_TECHNICAL_FAILURE
+from app.models import (
+    NOTIFICATION_FAILED,
+    NOTIFICATION_SENT,
+    NOTIFICATION_TECHNICAL_FAILURE,
+)
+
+
+@notify_celery.task(bind=True, name="check_sms_delivery_receipt", max_retries=48, default_retry_delay=300)
+def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
+    """
+    This is called after deliver_sms to check the status of the message. This uses the same number of
+    retries and the same delay period as deliver_sms.  In addition, this fires five minutes after
+    deliver_sms initially. So the idea is that most messages will succeed and show up in the logs quickly.
+    Other message will resolve successfully after a retry or to.  A few will fail but it will take up to
+    4 hours to know for sure.  The call to check_sms will raise an exception if neither a success nor a
+    failure appears in the cloudwatch logs, so this should keep retrying until the log appears, or until
+    we run out of retries.
+    """
+    status, provider_response = aws_cloudwatch_client.check_sms(message_id, notification_id, sent_at)
+    if status == 'success':
+        status = NOTIFICATION_SENT
+    else:
+        status = NOTIFICATION_FAILED
+    update_notification_status_by_id(notification_id, status, provider_response=provider_response)
+    current_app.logger.info(f"Updated notification {notification_id} with response '{provider_response}'")
 
 
 @notify_celery.task(bind=True, name="deliver_sms", max_retries=48, default_retry_delay=300)
 def deliver_sms(self, notification_id):
     try:
+        # Get the time we are doing the sending, to minimize the time period we need to check over for receipt
+        now = round(time() * 1000)
         current_app.logger.info("Start sending SMS for notification id: {}".format(notification_id))
         notification = notifications_dao.get_notification_by_id(notification_id)
         if not notification:
             raise NoResultFound()
-        send_to_providers.send_sms_to_provider(notification)
+        message_id = send_to_providers.send_sms_to_provider(notification)
+        # We have to put it in the default US/Eastern timezone.  From zones west of there, the delay
+        # will be ignored and it will fire immediately (although this probably only affects developer testing)
+        my_eta = datetime.now(ZoneInfo('US/Eastern')) + timedelta(seconds=300)
+        check_sms_delivery_receipt.apply_async(
+            [message_id, notification_id, now],
+            eta=my_eta,
+            queue=QueueNames.CHECK_SMS
+        )
     except Exception as e:
         if isinstance(e, SmsClientResponseException):
             current_app.logger.warning(

app/celery/research_mode_tasks.py

@@ -5,6 +5,7 @@ from requests import HTTPError, request
 
 from app.celery.process_ses_receipts_tasks import process_ses_results
 from app.config import QueueNames
+from app.dao.notifications_dao import get_notification_by_id
 from app.models import SMS_TYPE
 
 temp_fail = "2028675303"
@@ -16,8 +17,8 @@ perm_fail_email = "perm-fail@simulator.notify"
 temp_fail_email = "temp-fail@simulator.notify"
 
 
-def send_sms_response(provider, reference, to):
-    body = sns_callback(reference, to)
+def send_sms_response(provider, reference):
+    body = sns_callback(reference)
     headers = {"Content-type": "application/json"}
 
     make_request(SMS_TYPE, provider, body, headers)
@@ -59,25 +60,16 @@ def make_request(notification_type, provider, data, headers):
     return response.json()
 
 
-def sns_callback(notification_id, to):
-    raise Exception("Need to update for SNS callback format along with test_send_to_providers")
+def sns_callback(notification_id):
+    notification = get_notification_by_id(notification_id)
 
-    # example from mmg_callback
-    # if to.strip().endswith(temp_fail):
-    #     # status: 4 - expired (temp failure)
-    #     status = "4"
-    # elif to.strip().endswith(perm_fail):
-    #     # status: 5 - rejected (perm failure)
-    #     status = "5"
-    # else:
-    #     # status: 3 - delivered
-    #     status = "3"
-
-    # return json.dumps({"reference": "mmg_reference",
-    #                    "CID": str(notification_id),
-    #                    "MSISDN": to,
-    #                    "status": status,
-    #                    "deliverytime": "2016-04-05 16:01:07"})
+    # This will only work if all notifications, including successful ones, are in the notifications table
+    # If we decide to delete successful notifications, we will have to get this from notifications history
+    return json.dumps({
+        "CID": str(notification_id),
+        "status": notification.status,
+        # "deliverytime": notification.completed_at
+    })
 
 
 def ses_notification_callback(reference):