darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-22 08:21:13 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
248f5a07082e8ca620062bd59b08334bbbe7b74b
notifications-api/app/celery/celery.py

87 lines
3.1 KiB
Python
Raw Normal View History

Updated utils to the latest version. This version of utils has less logging at info level and as such no longer prints out the celery task timing which are found to be use to find out if a tasks has been called but also the timing for the task. Added an extra timing message for celery tasks so that it can be determined if the these are less frequent than the API calls and provide more useful information
2018-02-05 14:58:02 +00:00
import time
Run auto-correct on app/ and tests/
2021-03-10 13:55:06 +00:00
log unhandled celery exceptions they were always caught locally by celery's base handler, however, we weren't logging them ourselves, which meant it wouldn't be put into the json logs that are sent to cloudwatch.
2017-08-31 12:52:06 +01:00
from celery import Celery, Task
remove duplication shutdown loggers also add **kwargs to make it celery4 compatible
2018-07-12 15:09:38 +01:00
from celery.signals import worker_process_shutdown
restart reporting worker after each task The reporting worker tasks fetch large amounts of data from the db, do some processing then store back in the database. As the reporting worker only processes the create nightly billing/stats table tasks, which aren't high performance or high volume, we're fine with the performance hit from restarting the worker between every task (which based on limited local testing takes about a second or so). This causes some real funky shit with the app_context (used for accessing current_app.logger). To access flask's global state we use the standard way of importing `from flask import current_app`. However, processes after the first one don't have the current_app available on shut down (they're fine during the actual task running), and are unable to call `with current_app.app_context()` to create it. They _are_ able to call `with app.app_context()` to create it, where `app` is the initial app that we pass in to `NotifyCelery.init_app`. NotifyCelery.init_app is only called once, in the master process - I think the application state is then stored and passed to the celery workers. But then it looks like the teardown might clear it, but it never gets set up again for the new workers? Unsure. To fix this, store a copy of the initial flask app on the NotifyCelery object and then use that from within the shutdown signal logging function. Nothing's ever easy ¯\_(ツ)_/¯
2020-04-24 11:21:41 +01:00
from flask import g, request
Run auto-correct on app/ and tests/
2021-03-10 13:55:06 +00:00
from flask.ctx import has_app_context, has_request_context
remove duplication shutdown loggers also add **kwargs to make it celery4 compatible
2018-07-12 15:09:38 +01:00
@worker_process_shutdown.connect
restart reporting worker after each task The reporting worker tasks fetch large amounts of data from the db, do some processing then store back in the database. As the reporting worker only processes the create nightly billing/stats table tasks, which aren't high performance or high volume, we're fine with the performance hit from restarting the worker between every task (which based on limited local testing takes about a second or so). This causes some real funky shit with the app_context (used for accessing current_app.logger). To access flask's global state we use the standard way of importing `from flask import current_app`. However, processes after the first one don't have the current_app available on shut down (they're fine during the actual task running), and are unable to call `with current_app.app_context()` to create it. They _are_ able to call `with app.app_context()` to create it, where `app` is the initial app that we pass in to `NotifyCelery.init_app`. NotifyCelery.init_app is only called once, in the master process - I think the application state is then stored and passed to the celery workers. But then it looks like the teardown might clear it, but it never gets set up again for the new workers? Unsure. To fix this, store a copy of the initial flask app on the NotifyCelery object and then use that from within the shutdown signal logging function. Nothing's ever easy ¯\_(ツ)_/¯
2020-04-24 11:21:41 +01:00
def log_on_worker_shutdown(sender, signal, pid, exitcode, **kwargs):
# imported here to avoid circular imports
from app import notify_celery
# if the worker has already restarted at least once, then we no longer have app context and current_app won't work
# to create a new one. Instead we have to create a new app context from the original flask app and use that instead.
with notify_celery._app.app_context():
# if the worker has restarted
notify_celery._app.logger.info('worker shutdown: PID: {} Exitcode: {}'.format(pid, exitcode))
log unhandled celery exceptions they were always caught locally by celery's base handler, however, we weren't logging them ourselves, which meant it wouldn't be put into the json logs that are sent to cloudwatch.
2017-08-31 12:52:06 +01:00
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
def make_task(app):
class NotifyTask(Task):
abstract = True
start = None
Updated utils to the latest version. This version of utils has less logging at info level and as such no longer prints out the celery task timing which are found to be use to find out if a tasks has been called but also the timing for the task. Added an extra timing message for celery tasks so that it can be determined if the these are less frequent than the API calls and provide more useful information
2018-02-05 14:58:02 +00:00
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
def on_success(self, retval, task_id, args, kwargs):
Switch to monotonic time for task logs This matches the approach we take in utils [1]. Monotonic time is better because it avoids weird negative results due to clock shift. [1]: https://github.com/alphagov/notifications-utils/blob/5d18ebd79623e7d16bd48fbfe78ae632c0c0aaf1/notifications_utils/statsd_decorators.py#L14
2021-04-07 13:56:10 +01:00
elapsed_time = time.monotonic() - self.start
Include queue name in Celery task logs This is mainly so we can use it in the new metrics we send to StatsD in the following commits, but it should also be useful in the logs. I've taken the opportunity to make the log format consistent between success / failure, and with our Template Preview app [1]. [1]: https://github.com/alphagov/notifications-template-preview/blob/f456433a5ae7b41661ee014b6589bb31cc2e2c66/app/celery/celery.py#L19
2021-04-07 14:10:39 +01:00
delivery_info = self.request.delivery_info or {}
queue_name = delivery_info.get('routing_key', 'none')
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
app.logger.info(
Include queue name in Celery task logs This is mainly so we can use it in the new metrics we send to StatsD in the following commits, but it should also be useful in the logs. I've taken the opportunity to make the log format consistent between success / failure, and with our Template Preview app [1]. [1]: https://github.com/alphagov/notifications-template-preview/blob/f456433a5ae7b41661ee014b6589bb31cc2e2c66/app/celery/celery.py#L19
2021-04-07 14:10:39 +01:00
"Celery task {task_name} (queue: {queue_name}) took {time}".format(
task_name=self.name,
queue_name=queue_name,
time="{0:.4f}".format(elapsed_time)
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
)
Updated utils to the latest version. This version of utils has less logging at info level and as such no longer prints out the celery task timing which are found to be use to find out if a tasks has been called but also the timing for the task. Added an extra timing message for celery tasks so that it can be determined if the these are less frequent than the API calls and provide more useful information
2018-02-05 14:58:02 +00:00
)
log unhandled celery exceptions they were always caught locally by celery's base handler, however, we weren't logging them ourselves, which meant it wouldn't be put into the json logs that are sent to cloudwatch.
2017-08-31 12:52:06 +01:00
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
def on_failure(self, exc, task_id, args, kwargs, einfo):
Include queue name in Celery task logs This is mainly so we can use it in the new metrics we send to StatsD in the following commits, but it should also be useful in the logs. I've taken the opportunity to make the log format consistent between success / failure, and with our Template Preview app [1]. [1]: https://github.com/alphagov/notifications-template-preview/blob/f456433a5ae7b41661ee014b6589bb31cc2e2c66/app/celery/celery.py#L19
2021-04-07 14:10:39 +01:00
delivery_info = self.request.delivery_info or {}
queue_name = delivery_info.get('routing_key', 'none')
app.logger.exception(
"Celery task {task_name} (queue: {queue_name}) failed".format(
task_name=self.name,
queue_name=queue_name,
)
)
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
super().on_failure(exc, task_id, args, kwargs, einfo)
def __call__(self, *args, **kwargs):
# ensure task has flask context to access config, logger, etc
with app.app_context():
Switch to monotonic time for task logs This matches the approach we take in utils [1]. Monotonic time is better because it avoids weird negative results due to clock shift. [1]: https://github.com/alphagov/notifications-utils/blob/5d18ebd79623e7d16bd48fbfe78ae632c0c0aaf1/notifications_utils/statsd_decorators.py#L14
2021-04-07 13:56:10 +01:00
self.start = time.monotonic()
Include queue name in Celery task logs This is mainly so we can use it in the new metrics we send to StatsD in the following commits, but it should also be useful in the logs. I've taken the opportunity to make the log format consistent between success / failure, and with our Template Preview app [1]. [1]: https://github.com/alphagov/notifications-template-preview/blob/f456433a5ae7b41661ee014b6589bb31cc2e2c66/app/celery/celery.py#L19
2021-04-07 14:10:39 +01:00
# Remove piggyback values from kwargs
# Add 'request_id' to 'g' so that it gets logged
Pass request_id to tasks if available We want to pass the `request_id` to Celery tasks if the task is called from an HTTP request, so that we can add the `request_id` to the logs. This change overwrites `apply_async` to add the `request_id` to the kwargs if available. When we call the task, we then add the `request_id` to g on Flask's application context. Tasks called from `send_task` won't have a `request_id` for now, and this change only affects tasks called from HTTP requests (not from other tasks or from Celery Beat).
2019-10-23 16:14:59 +01:00
g.request_id = kwargs.pop('request_id', None)
Include queue name in Celery task logs This is mainly so we can use it in the new metrics we send to StatsD in the following commits, but it should also be useful in the logs. I've taken the opportunity to make the log format consistent between success / failure, and with our Template Preview app [1]. [1]: https://github.com/alphagov/notifications-template-preview/blob/f456433a5ae7b41661ee014b6589bb31cc2e2c66/app/celery/celery.py#L19
2021-04-07 14:10:39 +01:00
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
return super().__call__(*args, **kwargs)
log unhandled celery exceptions they were always caught locally by celery's base handler, however, we weren't logging them ourselves, which meant it wouldn't be put into the json logs that are sent to cloudwatch.
2017-08-31 12:52:06 +01:00
Pass request_id to tasks if available We want to pass the `request_id` to Celery tasks if the task is called from an HTTP request, so that we can add the `request_id` to the logs. This change overwrites `apply_async` to add the `request_id` to the kwargs if available. When we call the task, we then add the `request_id` to g on Flask's application context. Tasks called from `send_task` won't have a `request_id` for now, and this change only affects tasks called from HTTP requests (not from other tasks or from Celery Beat).
2019-10-23 16:14:59 +01:00
def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
link=None, link_error=None, **options):
kwargs = kwargs or {}
Include queue name in Celery task logs This is mainly so we can use it in the new metrics we send to StatsD in the following commits, but it should also be useful in the logs. I've taken the opportunity to make the log format consistent between success / failure, and with our Template Preview app [1]. [1]: https://github.com/alphagov/notifications-template-preview/blob/f456433a5ae7b41661ee014b6589bb31cc2e2c66/app/celery/celery.py#L19
2021-04-07 14:10:39 +01:00
Pass request_id to tasks if available We want to pass the `request_id` to Celery tasks if the task is called from an HTTP request, so that we can add the `request_id` to the logs. This change overwrites `apply_async` to add the `request_id` to the kwargs if available. When we call the task, we then add the `request_id` to g on Flask's application context. Tasks called from `send_task` won't have a `request_id` for now, and this change only affects tasks called from HTTP requests (not from other tasks or from Celery Beat).
2019-10-23 16:14:59 +01:00
if has_request_context() and hasattr(request, 'request_id'):
kwargs['request_id'] = request.request_id
Check for app_context and request in g to prevent Attribute Errors. We can add a request_id for tasks that are not spawned by an HTTP request, for example scheduled or nightly tasks. That means you can match up all the tasks spawned by a single task, for example, create-night-billing spawns 4 tasks, those would all have the same idea. Not sure if that is helpful or not. Also it might be confusing to have a request_id for logs that were not started from a request so I have left it out.
2020-12-22 15:47:35 +00:00
elif has_app_context() and 'request_id' in g:
If the request_id exists in the Flask global context, add it to the kwargs for the task. The request_id is set is the task is created from a http request, if that task then calls through to another task this will set the request_id from the global context. We should then be able to follow the creation of a notification all the way from the original http request to the sending task.
2020-12-21 14:53:14 +00:00
kwargs['request_id'] = g.request_id
Pass request_id to tasks if available We want to pass the `request_id` to Celery tasks if the task is called from an HTTP request, so that we can add the `request_id` to the logs. This change overwrites `apply_async` to add the `request_id` to the kwargs if available. When we call the task, we then add the `request_id` to g on Flask's application context. Tasks called from `send_task` won't have a `request_id` for now, and this change only affects tasks called from HTTP requests (not from other tasks or from Celery Beat).
2019-10-23 16:14:59 +01:00
Remove unused metric for SQS apply duration This was added as part of a wider performance investigation [1]. I checked with Leo, who made the change, and while the other metrics are still be useful, there's no reason to keep this one. [1]: https://github.com/alphagov/notifications-api/commit/6e32ca59963b53a8a404b9362a56fc689d6fd4c2#diff-76936416943346b5f691dac57a64acebc6a1227293820d1d9af4791087c9fb9eR23
2021-04-07 13:54:04 +01:00
return super().apply_async(args, kwargs, task_id, producer, link, link_error, **options)
Pass request_id to tasks if available We want to pass the `request_id` to Celery tasks if the task is called from an HTTP request, so that we can add the `request_id` to the logs. This change overwrites `apply_async` to add the `request_id` to the kwargs if available. When we call the task, we then add the `request_id` to g on Flask's application context. Tasks called from `send_task` won't have a `request_id` for now, and this change only affects tasks called from HTTP requests (not from other tasks or from Celery Beat).
2019-10-23 16:14:59 +01:00
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
return NotifyTask
hacked celery4.0.2 in. Runs and works - note though this version of master I branched had split head on sqlalchemey. This needs a new master merge to fix
2017-06-01 14:32:19 +01:00
Celery tests
2016-02-09 13:31:45 +00:00
class NotifyCelery(Celery):
Finished celery refactor - set up config for queue prefix LEO notes: Also made sure the Test BROKER_URL is preserved so that tests warn you when celery isn't mocked out
2017-06-09 16:20:02 +01:00
Revert celery4 Revert the following three pull requests: https://github.com/alphagov/notifications-api/pull/1085 https://github.com/alphagov/notifications-api/pull/1086 https://github.com/alphagov/notifications-api/pull/1088 celery 4.0.2 looked promising, however, on staging under mild load (5/sec api calls) the performance was actually worse than 3.1.25
2017-07-19 13:50:29 +01:00
def init_app(self, app):
log unhandled celery exceptions they were always caught locally by celery's base handler, however, we weren't logging them ourselves, which meant it wouldn't be put into the json logs that are sent to cloudwatch.
2017-08-31 12:52:06 +01:00
super().__init__(
app.import_name,
broker=app.config['BROKER_URL'],
Create an app context for each celery task run Celery tasks require an active app context in order to use app logger, database or app configuration values. Since there's no builtin support we create this context by using a custom celery task class NotifyTask. However, NotifyTask was using `current_app`, which itself is only available within an app context so the code was pushing an initial context in run_celery.py. This works with the default prefork pool implementation, but raises a "working outside of application context" error with an eventlet pool since that initial application context is local to a thread and eventlet celery worker pool will create multiple green threads. To avoid this, we bind NotifyTask to the app variable with a closure and use that variable instead of `current_app` to create the context for executing the task. This avoids any issues caused by shared initial app context being lost when spawning additional worker threads. We still need to keep the context push in run_celery.py for prefork workers since it's required for logging events outside of tasks (eg logging during `worker_process_shutdown` signal processing).
2018-02-12 15:29:03 +00:00
task_cls=make_task(app),
log unhandled celery exceptions they were always caught locally by celery's base handler, however, we weren't logging them ourselves, which meant it wouldn't be put into the json logs that are sent to cloudwatch.
2017-08-31 12:52:06 +01:00
)
Celery tests
2016-02-09 13:31:45 +00:00
log unhandled celery exceptions they were always caught locally by celery's base handler, however, we weren't logging them ourselves, which meant it wouldn't be put into the json logs that are sent to cloudwatch.
2017-08-31 12:52:06 +01:00
self.conf.update(app.config)
restart reporting worker after each task The reporting worker tasks fetch large amounts of data from the db, do some processing then store back in the database. As the reporting worker only processes the create nightly billing/stats table tasks, which aren't high performance or high volume, we're fine with the performance hit from restarting the worker between every task (which based on limited local testing takes about a second or so). This causes some real funky shit with the app_context (used for accessing current_app.logger). To access flask's global state we use the standard way of importing `from flask import current_app`. However, processes after the first one don't have the current_app available on shut down (they're fine during the actual task running), and are unable to call `with current_app.app_context()` to create it. They _are_ able to call `with app.app_context()` to create it, where `app` is the initial app that we pass in to `NotifyCelery.init_app`. NotifyCelery.init_app is only called once, in the master process - I think the application state is then stored and passed to the celery workers. But then it looks like the teardown might clear it, but it never gets set up again for the new workers? Unsure. To fix this, store a copy of the initial flask app on the NotifyCelery object and then use that from within the shutdown signal logging function. Nothing's ever easy ¯\_(ツ)_/¯
2020-04-24 11:21:41 +01:00
self._app = app
Reference in New Issue Copy Permalink