Merge pull request #5305 from AlanCoding/celery_dynamic_option

Get celery logging autoreload + don't reconfigure the whole config

awx/conf/apps.py

@@ -2,7 +2,7 @@
 from django.apps import AppConfig
 # from django.core import checks
 from django.utils.translation import ugettext_lazy as _
-from django.utils.log import configure_logging
+from awx.main.utils.handlers import configure_external_logger
 from django.conf import settings
 
 
@@ -15,10 +15,4 @@ class ConfConfig(AppConfig):
         self.module.autodiscover()
         from .settings import SettingsWrapper
         SettingsWrapper.initialize()
-        if settings.LOG_AGGREGATOR_ENABLED:
+        configure_external_logger(settings)
-            LOGGING_DICT = settings.LOGGING
-            LOGGING_DICT['handlers']['http_receiver']['class'] = 'awx.main.utils.handlers.HTTPSHandler'
-            if 'awx' in settings.LOG_AGGREGATOR_LOGGERS:
-                if 'http_receiver' not in LOGGING_DICT['loggers']['awx']['handlers']:
-                    LOGGING_DICT['loggers']['awx']['handlers'] += ['http_receiver']
-            configure_logging(settings.LOGGING_CONFIG, LOGGING_DICT)

awx/main/tasks.py

@@ -32,7 +32,7 @@ import pexpect
 
 # Celery
 from celery import Task, task
-from celery.signals import celeryd_init, worker_ready
+from celery.signals import celeryd_init, worker_process_init
 from celery import current_app
 
 # Django
@@ -54,6 +54,7 @@ from awx.main.task_engine import TaskEnhancer
 from awx.main.utils import (get_ansible_version, get_ssh_version, decrypt_field, update_scm_url,
                             check_proot_installed, build_proot_temp_dir, wrap_args_with_proot,
                             get_system_task_capacity, OutputEventFilter, parse_yaml_or_json)
+from awx.main.utils.handlers import configure_external_logger
 from awx.main.consumers import emit_channel_notification
 
 __all__ = ['RunJob', 'RunSystemJob', 'RunProjectUpdate', 'RunInventoryUpdate',
@@ -86,26 +87,10 @@ def celery_startup(conf=None, **kwargs):
             logger.error("Failed to rebuild schedule {}: {}".format(sch, e))
 
 
-def _setup_tower_logger():
+@worker_process_init.connect
-    global logger
-    from django.utils.log import configure_logging
-    LOGGING_DICT = settings.LOGGING
-    if settings.LOG_AGGREGATOR_ENABLED:
-        LOGGING_DICT['handlers']['http_receiver']['class'] = 'awx.main.utils.handlers.HTTPSHandler'
-        LOGGING_DICT['handlers']['http_receiver']['async'] = False
-        if 'awx' in settings.LOG_AGGREGATOR_LOGGERS:
-            if 'http_receiver' not in LOGGING_DICT['loggers']['awx']['handlers']:
-                LOGGING_DICT['loggers']['awx']['handlers'] += ['http_receiver']
-    configure_logging(settings.LOGGING_CONFIG, LOGGING_DICT)
-    logger = logging.getLogger('awx.main.tasks')
-
-
-@worker_ready.connect
 def task_set_logger_pre_run(*args, **kwargs):
     cache.close()
-    if settings.LOG_AGGREGATOR_ENABLED:
+    configure_external_logger(settings, async_flag=False, is_startup=False)
-        _setup_tower_logger()
-        logger.debug('Custom Tower logger configured for worker process.')
 
 
 def _uwsgi_reload():
@@ -121,7 +106,7 @@ def _uwsgi_reload():
 
 
 def _reset_celery_logging():
-    # Worker logger reloaded, now send signal to restart pool
+    # Send signal to restart thread pool
     app = current_app._get_current_object()
     app.control.broadcast('pool_restart', arguments={'reload': True},
                           destination=['celery@{}'.format(settings.CLUSTER_HOST_ID)], reply=False)

awx/main/utils/handlers.py

@@ -12,8 +12,11 @@ import traceback
 
 from requests_futures.sessions import FuturesSession
 
-# custom
+# AWX
-from django.conf import settings as django_settings
+from awx.main.utils.formatters import LogstashFormatter
+
+
+__all__ = ['HTTPSNullHandler', 'BaseHTTPSHandler', 'configure_external_logger']
 
 # AWX external logging handler, generally designed to be used
 # with the accompanying LogstashHandler, derives from python-logstash library
@@ -40,7 +43,7 @@ def unused_callback(sess, resp):
 class HTTPSNullHandler(logging.NullHandler):
     "Placeholder null handler to allow loading without database access"
 
-    def __init__(self, host, **kwargs):
+    def __init__(self, *args, **kwargs):
         return super(HTTPSNullHandler, self).__init__()
 
 
@@ -165,7 +168,31 @@ class BaseHTTPSHandler(logging.Handler):
                                  **self.get_post_kwargs(payload))
 
 
-class HTTPSHandler(object):
+def add_or_remove_logger(address, instance):
+    specific_logger = logging.getLogger(address)
+    for i, handler in enumerate(specific_logger.handlers):
+        if isinstance(handler, (HTTPSNullHandler, BaseHTTPSHandler)):
+            specific_logger.handlers[i] = instance or HTTPSNullHandler()
+            break
+    else:
+        if instance is not None:
+            specific_logger.handlers.append(instance)
 
-    def __new__(cls, *args, **kwargs):
+
-        return BaseHTTPSHandler.from_django_settings(django_settings, *args, **kwargs)
+def configure_external_logger(settings_module, async_flag=True, is_startup=True):
+
+    is_enabled = settings_module.LOG_AGGREGATOR_ENABLED
+    if is_startup and (not is_enabled):
+        # Pass-through if external logging not being used
+        return
+
+    instance = None
+    if is_enabled:
+        instance = BaseHTTPSHandler.from_django_settings(settings_module, async=async_flag)
+        instance.setFormatter(LogstashFormatter())
+    awx_logger_instance = instance
+    if is_enabled and 'awx' not in settings_module.LOG_AGGREGATOR_LOGGERS:
+        awx_logger_instance = None
+
+    add_or_remove_logger('awx.analytics', instance)
+    add_or_remove_logger('awx', awx_logger_instance)