Merge pull request #13438 from jessicamack/13437-cache-clear

[web-task-split] Rework django caching mechanism

Makefile

@@ -68,6 +68,12 @@ I18N_FLAG_FILE = .i18n_built
 	VERSION PYTHON_VERSION docker-compose-sources \
 	.git/hooks/pre-commit github_ci_setup github_ci_runner
 
+cache-clear:
+	@if [ "$(VENV_BASE)" ]; then \
+		. $(VENV_BASE)/awx/bin/activate; \
+	fi; \
+	$(PYTHON) manage.py run_cache_clear
+
 clean-tmp:
 	rm -rf tmp/
 

awx/main/dispatch/worker/task.py

@@ -26,8 +26,8 @@ class TaskWorker(BaseWorker):
     `awx.main.dispatch.publish`.
     """
 
-    @classmethod
-    def resolve_callable(cls, task):
+    @staticmethod
+    def resolve_callable(task):
         """
         Transform a dotted notation task into an imported, callable function, e.g.,
 
@@ -46,7 +46,8 @@ class TaskWorker(BaseWorker):
 
         return _call
 
-    def run_callable(self, body):
+    @staticmethod
+    def run_callable(body):
         """
         Given some AMQP message, import the correct Python code and run it.
         """

awx/main/management/commands/run_cache_clear.py

@@ -0,0 +1,32 @@
+import logging
+import json
+
+from django.core.management.base import BaseCommand
+from awx.main.dispatch import pg_bus_conn
+from awx.main.dispatch.worker.task import TaskWorker
+
+logger = logging.getLogger('awx.main.cache_clear')
+
+
+class Command(BaseCommand):
+    """
+    Cache Clear
+    Runs as a management command and starts a daemon that listens for a pg_notify message to clear the cache.
+    """
+
+    help = 'Launch the cache clear daemon'
+
+    def handle(self, *arg, **options):
+        try:
+            with pg_bus_conn(new_connection=True) as conn:
+                conn.listen("tower_settings_change")
+                for e in conn.events(yield_timeouts=True):
+                    if e is not None:
+                        body = json.loads(e.payload)
+                        logger.info(f"Cache clear request received. Clearing now, payload: {e.payload}")
+                        TaskWorker.run_callable(body)
+
+        except Exception:
+            # Log unanticipated exception in addition to writing to stderr to get timestamps and other metadata
+            logger.exception('Encountered unhandled error in cache clear main loop')
+            raise

awx/main/management/commands/run_dispatcher.py

@@ -76,7 +76,7 @@ class Command(BaseCommand):
         consumer = None
 
         try:
-            queues = ['tower_broadcast_all', 'rsyslog_configurer', get_local_queuename()]
+            queues = ['tower_broadcast_all', 'tower_settings_change', 'rsyslog_configurer', get_local_queuename()]
             consumer = AWXConsumerPG('dispatcher', TaskWorker(), queues, AutoscalePool(min_workers=4))
             consumer.run()
         except KeyboardInterrupt:

awx/main/tasks/system.py

@@ -61,6 +61,7 @@ from awx.main.utils.common import (
 
 from awx.main.utils.reload import stop_local_services
 from awx.main.utils.pglock import advisory_lock
+from awx.main.utils.external_logging import send_pg_notify
 from awx.main.tasks.receptor import get_receptor_ctl, worker_info, worker_cleanup, administrative_workunit_reaper, write_receptor_config
 from awx.main.consumers import emit_channel_notification
 from awx.main import analytics
@@ -240,8 +241,10 @@ def apply_cluster_membership_policies():
         logger.debug('Cluster policy computation finished in {} seconds'.format(time.time() - started_compute))
 
 
-@task(queue='tower_broadcast_all')
+@task(queue='tower_settings_change')
 def clear_setting_cache(setting_keys):
+    # log that cache is being cleared
+    logger.info(f"clear_setting_cache of keys {setting_keys}")
     orig_len = len(setting_keys)
     for i in range(orig_len):
         for dependent_key in settings_registry.get_dependent_settings(setting_keys[i]):

awx/settings/defaults.py

@@ -846,6 +846,7 @@ LOGGING = {
         'awx.main.consumers': {'handlers': ['console', 'file', 'tower_warnings'], 'level': 'INFO'},
         'awx.main.wsrelay': {'handlers': ['wsrelay']},
         'awx.main.rsyslog_configurer': {'handlers': ['rsyslog_configurer']},
+        'awx.main.cache_clear': {'handlers': ['cache_clear']},
         'awx.main.commands.inventory_import': {'handlers': ['inventory_import'], 'propagate': False},
         'awx.main.tasks': {'handlers': ['task_system', 'external_logger'], 'propagate': False},
         'awx.main.analytics': {'handlers': ['task_system', 'external_logger'], 'level': 'INFO', 'propagate': False},
@@ -877,6 +878,7 @@ handler_config = {
     'rbac_migrations': {'filename': 'tower_rbac_migrations.log'},
     'job_lifecycle': {'filename': 'job_lifecycle.log', 'formatter': 'job_lifecycle'},
     'rsyslog_configurer': {'filename': 'rsyslog_configurer.log'},
+    'cache_clear': {'filename': 'cache_clear.log'},
 }
 
 # If running on a VM, we log to files. When running in a container, we log to stdout.

tools/ansible/roles/dockerfile/templates/supervisor.conf.j2

@@ -74,8 +74,19 @@ stdout_logfile_maxbytes=0
 stderr_logfile=/dev/stderr
 stderr_logfile_maxbytes=0
 
+[program:awx-cache-clear]
+command = awx-manage run_cache_clear
+autorestart = true
+startsecs = 30
+stopasgroup=true
+killasgroup=true
+stdout_logfile=/dev/stdout
+stdout_logfile_maxbytes=0
+stderr_logfile=/dev/stderr
+stderr_logfile_maxbytes=0
+
 [group:tower-processes]
-programs=nginx,uwsgi,daphne
+programs=nginx,uwsgi,daphne,awx-cache-clear
 priority=5
 
 [eventlistener:superwatcher]

tools/docker-compose/supervisor.conf

@@ -47,6 +47,16 @@ stdout_logfile_maxbytes=0
 stderr_logfile=/dev/stderr
 stderr_logfile_maxbytes=0
 
+[program:awx-cache-clear]
+command = make cache-clear
+autorestart = true
+stopasgroup=true
+killasgroup=true
+stdout_logfile=/dev/stdout
+stdout_logfile_maxbytes=0
+stderr_logfile=/dev/stderr
+stderr_logfile_maxbytes=0
+
 [program:awx-uwsgi]
 command = make uwsgi
 autorestart = true