darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-20 23:41:17 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
3b0b96834d5aa6afc8399e6f49008b2483d950c2
notifications-api/app/celery/celery.py

79 lines
2.8 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
add prometheus metrics for connection pools (both web and sql) add the following prometheus metrics to keep track of general app instance health. # sqs_apply_async_duration how long does the actual SQS call (a standard web request to AWS) take. a histogram with default bucket sizes, split up per task that was created. # concurrent_web_request_count how many web requests is this app currently serving. this is split up per process, so we'd expect multiple responses per app instance # db_connection_total_connected how many connections does this app (process) have open to the database. They might be idle. # db_connection_total_checked_out how many connections does this app (process) have open that are currently in use by a web worker # db_connection_open_duration_seconds a histogram per endpoint of how long the db connection was taken from the pool for. won't have any data if a connection was never opened.
2020-04-24 14:36:21 +01:00
from gds_metrics.metrics import Histogram
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
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
from flask.ctx import 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):
dd sqlalchemy connection metrics for celery tasks grab the worker app name and task name rather than the web host and endpoint. also add a fallback for if we're not in a web request or a celery task. I think that'll probably happen when we use alembic, or if we do things from within flask shell
2020-04-24 15:15:41 +01:00
SQS_APPLY_ASYNC_DURATION_SECONDS = Histogram(
'sqs_apply_async_duration_seconds',
add prometheus metrics for connection pools (both web and sql) add the following prometheus metrics to keep track of general app instance health. # sqs_apply_async_duration how long does the actual SQS call (a standard web request to AWS) take. a histogram with default bucket sizes, split up per task that was created. # concurrent_web_request_count how many web requests is this app currently serving. this is split up per process, so we'd expect multiple responses per app instance # db_connection_total_connected how many connections does this app (process) have open to the database. They might be idle. # db_connection_total_checked_out how many connections does this app (process) have open that are currently in use by a web worker # db_connection_open_duration_seconds a histogram per endpoint of how long the db connection was taken from the pool for. won't have any data if a connection was never opened.
2020-04-24 14:36:21 +01:00
'Time taken to put task on queue',
['task_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
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):
elapsed_time = time.time() - self.start
app.logger.info(
"{task_name} took {time}".format(
task_name=self.name, time="{0:.4f}".format(elapsed_time)
)
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):
# ensure task will log exceptions to correct handlers
Added a new exception type for DVLAException. The Notify team needs to investigate when a notification is marked as failed. We will process the whole file and mark the notifications with the appropriate status, if any are failed an exception is raised. The exception will trigger a cloud watch error for the team to investigate.
2018-02-22 15:05:37 +00:00
app.logger.exception('Celery task: {} failed'.format(self.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():
self.start = time.time()
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 'request_id' from the kwargs (so the task doesn't get an unexpected kwarg), then add it to g
# so that it gets logged
g.request_id = kwargs.pop('request_id', 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
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 {}
if has_request_context() and hasattr(request, 'request_id'):
kwargs['request_id'] = request.request_id
dd sqlalchemy connection metrics for celery tasks grab the worker app name and task name rather than the web host and endpoint. also add a fallback for if we're not in a web request or a celery task. I think that'll probably happen when we use alembic, or if we do things from within flask shell
2020-04-24 15:15:41 +01:00
with SQS_APPLY_ASYNC_DURATION_SECONDS.labels(self.name).time():
add prometheus metrics for connection pools (both web and sql) add the following prometheus metrics to keep track of general app instance health. # sqs_apply_async_duration how long does the actual SQS call (a standard web request to AWS) take. a histogram with default bucket sizes, split up per task that was created. # concurrent_web_request_count how many web requests is this app currently serving. this is split up per process, so we'd expect multiple responses per app instance # db_connection_total_connected how many connections does this app (process) have open to the database. They might be idle. # db_connection_total_checked_out how many connections does this app (process) have open that are currently in use by a web worker # db_connection_open_duration_seconds a histogram per endpoint of how long the db connection was taken from the pool for. won't have any data if a connection was never opened.
2020-04-24 14:36:21 +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