Merge pull request #1310 from alphagov/rc_adding_logging_to_celery_tasks

Adding extra logging to celery tasks and gunicorn

app/celery/provider_tasks.py

@@ -1,3 +1,4 @@
+from celery.signals import worker_process_init, worker_process_shutdown, worker_shutdown
 from flask import current_app
 from notifications_utils.recipients import InvalidEmailError
 from sqlalchemy.orm.exc import NoResultFound
@@ -10,6 +11,11 @@ from app.statsd_decorators import statsd
 from app.delivery import send_to_providers
 
 
+@worker_process_shutdown.connect
+def worker_process_shutdown(sender, signal, pid, exitcode):
+    current_app.logger.info('Provider worker shutdown: PID: {} Exitcode: {}'.format(pid, exitcode))
+
+
 @notify_celery.task(bind=True, name="deliver_sms", max_retries=48, default_retry_delay=300)
 @statsd(namespace="tasks")
 def deliver_sms(self, notification_id):

app/celery/scheduled_tasks.py

@@ -3,6 +3,7 @@ from datetime import (
     timedelta
 )
 
+from celery.signals import worker_process_shutdown
 from flask import current_app
 from sqlalchemy.exc import SQLAlchemyError
 from notifications_utils.s3 import s3upload
@@ -45,6 +46,11 @@ from app.config import QueueNames, TaskNames
 from app.utils import convert_utc_to_bst
 
 
+@worker_process_shutdown.connect
+def worker_process_shutdown(sender, signal, pid, exitcode):
+    current_app.logger.info('Scheduled tasks worker shutdown: PID: {} Exitcode: {}'.format(pid, exitcode))
+
+
 @notify_celery.task(name="remove_csv_files")
 @statsd(namespace="tasks")
 def remove_csv_files(job_types):

app/celery/statistics_tasks.py

@@ -1,3 +1,4 @@
+from celery.signals import worker_process_shutdown
 from sqlalchemy.exc import SQLAlchemyError
 
 from app import notify_celery
@@ -23,6 +24,11 @@ def create_outcome_notification_statistic_tasks(notification):
         record_outcome_job_statistics.apply_async((str(notification.id),), queue=QueueNames.STATISTICS)
 
 
+@worker_process_shutdown.connect
+def worker_process_shutdown(sender, signal, pid, exitcode):
+    current_app.logger.info('Statistics worker shutdown: PID: {} Exitcode: {}'.format(pid, exitcode))
+
+
 @notify_celery.task(bind=True, name='record_initial_job_statistics', max_retries=20, default_retry_delay=10)
 @statsd(namespace="tasks")
 def record_initial_job_statistics(self, notification_id):

app/celery/tasks.py

@@ -2,6 +2,7 @@ import json
 from datetime import (datetime)
 from collections import namedtuple
 
+from celery.signals import worker_process_init, worker_process_shutdown
 from flask import current_app
 from notifications_utils.recipients import (
     RecipientCSV
@@ -56,6 +57,11 @@ from app.statsd_decorators import statsd
 from notifications_utils.s3 import s3upload
 
 
+@worker_process_shutdown.connect
+def worker_process_shutdown(sender, signal, pid, exitcode):
+    current_app.logger.info('Tasks worker shutdown: PID: {} Exitcode: {}'.format(pid, exitcode))
+
+
 @notify_celery.task(name="process-job")
 @statsd(namespace="tasks")
 def process_job(job_id):

gunicorn_config.py

@@ -2,7 +2,19 @@ import sys
 import traceback
 
 
+def on_starting(server):
+    server.log.info("Starting Notifications API")
+
+
 def worker_abort(worker):
-    worker.log.info("worker received ABORT")
+    worker.log.info("worker received ABORT {}".format(worker.pid))
     for threadId, stack in sys._current_frames().items():
         worker.log.error(''.join(traceback.format_stack(stack)))
+
+
+def on_exit(server):
+    server.log.info("Stopping Notifications API")
+
+
+def worker_int(worker):
+    worker.log.info("worker: received SIGINT {}".format(worker.pid))