darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-22 16:31:15 -05:00
Code Issues Packages Projects Releases Wiki Activity

Put the statistics tasks into a new queue. This currently has no worker in PaaS so blocks the deployment until new PaaS worker is configured.

Browse Source
This commit is contained in:
Martyn Inglis
2017-05-09 22:03:13 +01:00
parent caed193647
commit a4549a2e26
3 changed files with 16 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
app/celery/statistics_tasks.py
View File

@@ -12,17 +12,18 @@ from app.dao.notifications_dao import get_notification_by_id
def create_initial_notification_statistic_tasks(notification): def create_initial_notification_statistic_tasks(notification):
if notification.job_id: if notification.job_id:
record_initial_job_statistics.apply_async((str(notification.id),), queue="notify") record_initial_job_statistics.apply_async((str(notification.id),), queue="statistics")
def create_outcome_notification_statistic_tasks(notification): def create_outcome_notification_statistic_tasks(notification):
if notification.job_id: if notification.job_id:
record_outcome_job_statistics.apply_async((str(notification.id),), queue="notify") record_outcome_job_statistics.apply_async((str(notification.id),), queue="statistics")
@notify_celery.task(bind=True, name='record_initial_job_statistics', max_retries=20, default_retry_delay=10) @notify_celery.task(bind=True, name='record_initial_job_statistics', max_retries=20, default_retry_delay=10)
@statsd(namespace="tasks") @statsd(namespace="tasks")
def record_initial_job_statistics(self, notification_id): def record_initial_job_statistics(self, notification_id):
notification = None
try: try:
notification = get_notification_by_id(notification_id) notification = get_notification_by_id(notification_id)
if notification: if notification:
@@ -34,13 +35,16 @@ def record_initial_job_statistics(self, notification_id):
self.retry(queue="retry") self.retry(queue="retry")
except self.MaxRetriesExceededError: except self.MaxRetriesExceededError:
current_app.logger.error( current_app.logger.error(
"RETRY FAILED: task record_initial_job_statistics failed for notification {}".format(notification.id) "RETRY FAILED: task record_initial_job_statistics failed for notification {}".format(
notification.id if notification else "missing ID"
)
) )
@notify_celery.task(bind=True, name='record_outcome_job_statistics', max_retries=20, default_retry_delay=10) @notify_celery.task(bind=True, name='record_outcome_job_statistics', max_retries=20, default_retry_delay=10)
@statsd(namespace="tasks") @statsd(namespace="tasks")
def record_outcome_job_statistics(self, notification_id): def record_outcome_job_statistics(self, notification_id):
notification = None
try: try:
notification = get_notification_by_id(notification_id) notification = get_notification_by_id(notification_id)
if notification: if notification:
@@ -54,5 +58,7 @@ def record_outcome_job_statistics(self, notification_id):
self.retry(queue="retry") self.retry(queue="retry")
except self.MaxRetriesExceededError: except self.MaxRetriesExceededError:
current_app.logger.error( current_app.logger.error(
"RETRY FAILED: task update_job_stats_outcome_count failed for notification {}".format(notification.id) "RETRY FAILED: task update_job_stats_outcome_count failed for notification {}".format(
notification.id if notification else "missing ID"
)
) )

6
app/config.py
View File

@@ -211,7 +211,8 @@ class Development(Config):
Queue('db-letter', Exchange('default'), routing_key='db-letter'), Queue('db-letter', Exchange('default'), routing_key='db-letter'),
Queue('send-sms', Exchange('default'), routing_key='send-sms'), Queue('send-sms', Exchange('default'), routing_key='send-sms'),
Queue('send-email', Exchange('default'), routing_key='send-email'), Queue('send-email', Exchange('default'), routing_key='send-email'),
Queue('research-mode', Exchange('default'), routing_key='research-mode') Queue('research-mode', Exchange('default'), routing_key='research-mode'),
Queue('statistics', Exchange('default'), routing_key='statistics')
] ]
API_HOST_NAME = "http://localhost:6011" API_HOST_NAME = "http://localhost:6011"
API_RATE_LIMIT_ENABLED = True API_RATE_LIMIT_ENABLED = True
@@ -234,7 +235,8 @@ class Test(Config):
Queue('db-letter', Exchange('default'), routing_key='db-letter'), Queue('db-letter', Exchange('default'), routing_key='db-letter'),
Queue('send-sms', Exchange('default'), routing_key='send-sms'), Queue('send-sms', Exchange('default'), routing_key='send-sms'),
Queue('send-email', Exchange('default'), routing_key='send-email'), Queue('send-email', Exchange('default'), routing_key='send-email'),
Queue('research-mode', Exchange('default'), routing_key='research-mode') Queue('research-mode', Exchange('default'), routing_key='research-mode'),
Queue('statistics', Exchange('default'), routing_key='statistics')
] ]
REDIS_ENABLED = True REDIS_ENABLED = True
API_RATE_LIMIT_ENABLED = True API_RATE_LIMIT_ENABLED = True

5
tests/app/celery/test_statistics_tasks.py
View File

@@ -14,7 +14,7 @@ def test_should_create_initial_job_task_if_notification_is_related_to_a_job(
mock = mocker.patch("app.celery.statistics_tasks.record_initial_job_statistics.apply_async") mock = mocker.patch("app.celery.statistics_tasks.record_initial_job_statistics.apply_async")
notification = sample_notification(notify_db, notify_db_session, job=sample_job) notification = sample_notification(notify_db, notify_db_session, job=sample_job)
create_initial_notification_statistic_tasks(notification) create_initial_notification_statistic_tasks(notification)
mock.assert_called_once_with((str(notification.id), ), queue="notify") mock.assert_called_once_with((str(notification.id), ), queue="statistics")
def test_should_not_create_initial_job_task_if_notification_is_related_to_a_job( def test_should_not_create_initial_job_task_if_notification_is_related_to_a_job(
@@ -31,7 +31,7 @@ def test_should_create_outcome_job_task_if_notification_is_related_to_a_job(
mock = mocker.patch("app.celery.statistics_tasks.record_outcome_job_statistics.apply_async") mock = mocker.patch("app.celery.statistics_tasks.record_outcome_job_statistics.apply_async")
notification = sample_notification(notify_db, notify_db_session, job=sample_job) notification = sample_notification(notify_db, notify_db_session, job=sample_job)
create_outcome_notification_statistic_tasks(notification) create_outcome_notification_statistic_tasks(notification)
mock.assert_called_once_with((str(notification.id), ), queue="notify") mock.assert_called_once_with((str(notification.id), ), queue="statistics")
def test_should_not_create_outcome_job_task_if_notification_is_related_to_a_job( def test_should_not_create_outcome_job_task_if_notification_is_related_to_a_job(
@@ -42,7 +42,6 @@ def test_should_not_create_outcome_job_task_if_notification_is_related_to_a_job(
mock.assert_not_called() mock.assert_not_called()
def test_should_call_create_job_stats_dao_methods(notify_db, notify_db_session, sample_notification, mocker): def test_should_call_create_job_stats_dao_methods(notify_db, notify_db_session, sample_notification, mocker):
dao_mock = mocker.patch("app.celery.statistics_tasks.create_or_update_job_sending_statistics") dao_mock = mocker.patch("app.celery.statistics_tasks.create_or_update_job_sending_statistics")
record_initial_job_statistics(str(sample_notification.id)) record_initial_job_statistics(str(sample_notification.id))